merged fixes for BUG#39934 to 5.1-rpl+3

Also renamed current_stmt_binlog_row_based to
current_stmt_binlog_format_row for consistency

31 files changed, 961 insertions, 463 deletions

diff --git a/sql/event_db_repository.cc b/sql/event_db_repository.cc
index 9a253d74546..80aa93a0ca4 100644
--- a/sql/event_db_repository.cc
+++ b/sql/event_db_repository.cc
@@ -1034,8 +1034,8 @@ 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)
-    thd->clear_current_stmt_binlog_row_based();
+  if (thd->is_current_stmt_binlog_format_row())
+    thd->clear_current_stmt_binlog_format_row();
 
   DBUG_ASSERT(thd->security_ctx->master_access & SUPER_ACL);
 
diff --git a/sql/events.cc b/sql/events.cc
index 34da0e185b7..5b9acb589b7 100644
--- a/sql/events.cc
+++ b/sql/events.cc
@@ -430,8 +430,8 @@ 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)
-    thd->clear_current_stmt_binlog_row_based();
+  if (thd->is_current_stmt_binlog_format_row())
+    thd->clear_current_stmt_binlog_format_row();
 
   pthread_mutex_lock(&LOCK_event_metadata);
 
@@ -563,8 +563,8 @@ 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)
-    thd->clear_current_stmt_binlog_row_based();
+  if (thd->is_current_stmt_binlog_format_row())
+    thd->clear_current_stmt_binlog_format_row();
 
   pthread_mutex_lock(&LOCK_event_metadata);
 
@@ -660,8 +660,8 @@ 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)
-    thd->clear_current_stmt_binlog_row_based();
+  if (thd->is_current_stmt_binlog_format_row())
+    thd->clear_current_stmt_binlog_format_row();
 
   pthread_mutex_lock(&LOCK_event_metadata);
   /* On error conditions my_error() is called so no need to handle here */
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index d9a9738ce72..aba71b6f61d 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -302,6 +302,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_format_row();
 
   if (thd == injector_thd)
   {
@@ -3647,6 +3648,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_format_row();
 
   /* 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 ac8c46ec4e3..4b6ce28cb97 100644
--- a/sql/ha_partition.cc
+++ b/sql/ha_partition.cc
@@ -6195,7 +6195,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 e5c64452aaf..b44c799f499 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -2416,7 +2416,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);
@@ -4447,7 +4447,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 7991d9adf82..0d21c50c946 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -2375,10 +2375,11 @@ 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;
 
-  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));
@@ -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)
 {
-  thd->lex->set_stmt_unsafe();
+  DBUG_ENTER("Create_func_found_rows::create");
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   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)
 {
-  thd->lex->set_stmt_unsafe();
+  DBUG_ENTER("Create_func_load_file::create");
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   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)
 {
-  thd->lex->set_stmt_unsafe();
+  DBUG_ENTER("Create_func_row_count::create");
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   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)
 {
-  thd->lex->set_stmt_unsafe();
+  DBUG_ENTER("Create_func_uuid::create");
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   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)
 {
-  thd->lex->set_stmt_unsafe();
+  DBUG_ENTER("Create_func_uuid_short::create");
+  thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_SYSTEM_FUNCTION);
   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 1af2f3a4ddc..4c30ecb94c1 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -3837,7 +3837,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
@@ -4114,7 +4114,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 0cda724b698..2445560e65e 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -7223,7 +7223,7 @@ int Rows_log_event::do_apply_event(Relay_log_info const *rli)
 
       We also call the mysql_reset_thd_for_next_command(), since this
       is the logical start of the next "statement". Note that this
-      call might reset the value of current_stmt_binlog_row_based, so
+      call might reset the value of current_stmt_binlog_format, so
       we need to do any changes to that value after this function.
     */
     lex_start(thd);
@@ -7235,16 +7235,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.
@@ -7492,7 +7488,14 @@ 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);
-    thd->reset_current_stmt_binlog_row_based();
+    /*
+      @todo We should probably not call
+      reset_current_stmt_binlog_format_row() from here.
+
+      Note: this applies to log_event_old.cc too.
+      /Sven
+    */
+    thd->reset_current_stmt_binlog_format_row();
     const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
     thd->is_slave_error= 1;
   }
@@ -7593,7 +7596,17 @@ static int rows_event_stmt_cleanup(Relay_log_info const *rli, THD * thd)
       event flushed.
     */
 
-    thd->reset_current_stmt_binlog_row_based();
+    /*
+      @todo We should probably not call
+      reset_current_stmt_binlog_format_row() 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_format_row();
 
     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..339e1c443bf 100644
--- a/sql/log_event_old.cc
+++ b/sql/log_event_old.cc
@@ -59,22 +59,19 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info
 
       We also call the mysql_reset_thd_for_next_command(), since this
       is the logical start of the next "statement". Note that this
-      call might reset the value of current_stmt_binlog_row_based, so
+      call might reset the value of current_stmt_binlog_format, so
       we need to do any changes to that value after this function.
     */
     lex_start(thd);
     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))
     {
@@ -263,7 +260,7 @@ Old_rows_log_event::do_apply_event(Old_rows_log_event *ev, const Relay_log_info
       thread is certainly going to stop.
       rollback at the caller along with sbr.
     */
-    thd->reset_current_stmt_binlog_row_based();
+    thd->reset_current_stmt_binlog_format_row();
     const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
     thd->is_slave_error= 1;
     DBUG_RETURN(error);
@@ -1781,7 +1778,7 @@ int Old_rows_log_event::do_apply_event(Relay_log_info const *rli)
       thread is certainly going to stop.
       rollback at the caller along with sbr.
     */
-    thd->reset_current_stmt_binlog_row_based();
+    thd->reset_current_stmt_binlog_format_row();
     const_cast<Relay_log_info*>(rli)->cleanup_context(thd, error);
     thd->is_slave_error= 1;
     DBUG_RETURN(error);
@@ -1881,7 +1878,7 @@ Old_rows_log_event::do_update_pos(Relay_log_info *rli)
       event flushed.
     */
 
-    thd->reset_current_stmt_binlog_row_based();
+    thd->reset_current_stmt_binlog_format_row();
     rli->cleanup_context(thd, 0);
     if (error == 0)
     {
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 381a0313add..67242710fb2 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -1566,7 +1566,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..3a217066776 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1292,7 +1292,12 @@ bool sys_var_thd_binlog_format::is_readonly() const
 
 void fix_binlog_format_after_update(THD *thd, enum_var_type type)
 {
-  thd->reset_current_stmt_binlog_row_based();
+  /*
+    @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_format_row();
 }
 
 
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 3aba434b284..8c331b156c0 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
@@ -6206,3 +6205,38 @@ ER_TOO_MANY_CONCURRENT_TRXS
 
 WARN_NON_ASCII_SEPARATOR_NOT_IMPLEMENTED
   eng "Non-ASCII separator arguments are not fully supported"
+
+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."
+
+ER_BINLOG_UNSAFE_LIMIT
+  eng "Statement uses a LIMIT clause. This is unsafe because the set of rows included cannot be predicted."
+ER_BINLOG_UNSAFE_INSERT_DELAYED
+  eng "Statement uses INSERT DELAYED. This is unsafe because the time when rows are inserted cannot be predicted."
+ER_BINLOG_UNSAFE_SYSTEM_TABLE
+  eng "Statement uses the general_log or slow_log table. This is unsafe because system tables may differ on slave."
+ER_BINLOG_UNSAFE_TWO_AUTOINC_COLUMNS
+  eng "Statement updates two AUTO_INCREMENT columns. This is unsafe because the generated value cannot be predicted by slave."
+ER_BINLOG_UNSAFE_UDF
+  eng "Statement uses a UDF. It cannot be determined if the UDF will return the same value on slave."
+ER_BINLOG_UNSAFE_SYSTEM_VARIABLE
+  eng "Statement uses a system variable whose value may differ on slave."
+ER_BINLOG_UNSAFE_SYSTEM_FUNCTION
+  eng "Statement uses a system function whose value may differ on slave."
+
+ER_BINLOG_UNSAFE_WARNING_SHORT
+  eng "%s Reason: %s"
+ER_BINLOG_UNSAFE_WARNING_LONG
+  eng "%s Reason: %s Statement: %s"
diff --git a/sql/sp.cc b/sql/sp.cc
index 4d840f53e2f..9fafe4ac355 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -744,7 +744,7 @@ sp_create_routine(THD *thd, int type, sp_head *sp)
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   saved_count_cuted_fields= thd->count_cuted_fields;
   thd->count_cuted_fields= CHECK_FIELD_WARN;
@@ -936,7 +936,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;
@@ -985,7 +985,7 @@ sp_drop_routine(THD *thd, int type, sp_name *name)
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   if (!(table= open_proc_table_for_update(thd)))
     DBUG_RETURN(SP_OPEN_TABLE_FAILED);
@@ -1039,7 +1039,7 @@ sp_update_routine(THD *thd, int type, sp_name *name, st_sp_chistics *chistics)
     row-based replication. The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   if (!(table= open_proc_table_for_update(thd)))
     DBUG_RETURN(SP_OPEN_TABLE_FAILED);
@@ -1827,6 +1827,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;
 
@@ -1860,7 +1862,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 0736e5fc2a8..bb871a41846 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 };
@@ -1691,7 +1691,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
@@ -2103,13 +2103,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 dd11f8693ac..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,20 +457,24 @@ 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)
   {
+    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
       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_acl.cc b/sql/sql_acl.cc
index ab18a2d1d04..10fefd5bb04 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:
@@ -3040,7 +3040,7 @@ int mysql_table_grant(THD *thd, TABLE_LIST *table_list,
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
 #ifdef HAVE_REPLICATION
   /*
@@ -3257,7 +3257,7 @@ bool mysql_routine_grant(THD *thd, TABLE_LIST *table_list, bool is_proc,
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
 #ifdef HAVE_REPLICATION
   /*
@@ -3395,7 +3395,7 @@ bool mysql_grant(THD *thd, const char *db, List <LEX_USER> &list,
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
 #ifdef HAVE_REPLICATION
   /*
@@ -5621,7 +5621,7 @@ bool mysql_create_user(THD *thd, List <LEX_USER> &list)
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   /* CREATE USER may be skipped on replication client. */
   if ((result= open_grant_tables(thd, tables)))
@@ -5701,7 +5701,7 @@ bool mysql_drop_user(THD *thd, List <LEX_USER> &list)
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   /* DROP USER may be skipped on replication client. */
   if ((result= open_grant_tables(thd, tables)))
@@ -5775,7 +5775,7 @@ bool mysql_rename_user(THD *thd, List <LEX_USER> &list)
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   /* RENAME USER may be skipped on replication client. */
   if ((result= open_grant_tables(thd, tables)))
@@ -5857,7 +5857,7 @@ bool mysql_revoke_all(THD *thd,  List <LEX_USER> &list)
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   if ((result= open_grant_tables(thd, tables)))
     DBUG_RETURN(result != 1);
@@ -6110,7 +6110,7 @@ bool sp_revoke_privileges(THD *thd, const char *sp_db, const char *sp_name,
     row-based replication.  The flag will be reset at the end of the
     statement.
   */
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
 
   /* Remove procedure access */
   do
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index f55d3fc5006..10185705efe 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -28,7 +28,6 @@
 #include <io.h>
 #endif
 
-#define FLAGSTR(S,F) ((S) & (F) ? #F " " : "")
 
 /**
   This internal handler is used to trap internally
@@ -1441,7 +1440,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)
@@ -4835,7 +4834,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().
 */
 
@@ -5055,156 +5054,6 @@ static void mark_real_tables_as_free_for_reuse(TABLE_LIST *table)
 }
 
 
-/**
-   Decide on logging format to use for the statement.
-
-   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.
-
-   We now have three alternatives that prevent the statement from
-   being loggable:
-
-   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.
-
-   2. Statement mode is set, but the capabilities indicate that
-      statement format is not possible.
-
-   3. Row mode is set, but the capabilities indicate that row
-      format is not possible.
-
-   4. Statement is unsafe, but the capabilities indicate that row
-      format is not possible.
-
-   If we are in MIXED mode, we then decide what logging format to use:
-
-   1. If the statement is unsafe, row-based logging is used.
-
-   2. If statement-based logging is not possible, row-based logging is
-      used.
-
-   3. Otherwise, statement-based logging is used.
-
-   @param thd    Client thread
-   @param tables Tables involved in the query
- */
-
-int decide_logging_format(THD *thd, TABLE_LIST *tables)
-{
-  if (mysql_bin_log.is_open() && (thd->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;
-    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();
-      if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
-      {
-        ulonglong const flags= table->table->file->ha_table_flags();
-        DBUG_PRINT("info", ("table: %s; ha_table_flags: %s%s",
-                            table->table_name,
-                            FLAGSTR(flags, HA_BINLOG_STMT_CAPABLE),
-                            FLAGSTR(flags, HA_BINLOG_ROW_CAPABLE)));
-        if (prev_ht && prev_ht != table->table->file->ht)
-          multi_engine= TRUE;
-        prev_ht= table->table->file->ht;
-        flags_all_set &= flags;
-        flags_some_set |= flags;
-      }
-    }
-
-    DBUG_PRINT("info", ("flags_all_set: %s%s",
-                        FLAGSTR(flags_all_set, HA_BINLOG_STMT_CAPABLE),
-                        FLAGSTR(flags_all_set, HA_BINLOG_ROW_CAPABLE)));
-    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", ("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");
-    }
-
-    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)
-    {
-      thd->set_current_stmt_binlog_row_based_if_mixed();
-    }
-  }
-
-  return 0;
-}
-
 /*
   Lock all tables in list
 
@@ -5246,7 +5095,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
@@ -5284,8 +5133,8 @@ 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->set_current_stmt_binlog_row_based_if_mixed();
+        thd->lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_TWO_AUTOINC_COLUMNS);
+        thd->set_current_stmt_binlog_format_row_if_mixed();
       }
     }
 
@@ -5406,7 +5255,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 3f568566c89..46f05441a44 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_table_maps(0), binlog_flags(0UL),
+   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),
@@ -804,7 +805,7 @@ void THD::init(void)
   bzero((char*) warn_count, sizeof(warn_count));
   total_warn_count= 0;
   update_charset();
-  reset_current_stmt_binlog_row_based();
+  reset_current_stmt_binlog_format_row();
   bzero((char *) &status_var, sizeof(status_var));
   sql_log_bin_toplevel= options & OPTION_BIN_LOG;
 }
@@ -3041,13 +3042,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 */
@@ -3109,7 +3110,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);
 
   /*
@@ -3242,6 +3243,311 @@ void xid_cache_delete(XID_STATE *xid_state)
   pthread_mutex_unlock(&LOCK_xid_cache);
 }
 
+
+/**
+  Decide on logging format to use for the statement and issue errors
+  or warnings as needed.  The decision depends on the following
+  parameters:
+
+  - The logging mode, i.e., the value of binlog_format.  Can be
+    statement, mixed, or row.
+
+  - 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.
+
+  - 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*.
+
+    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.
+
+  Decision table for logging format
+  ---------------------------------
+
+  The following table summarizes how the format and generated
+  warning/error depends on the tables' capabilities, the statement
+  type, and the current binlog_format.
+
+     Row capable        N NNNNNNNNN YYYYYYYYY YYYYYYYYY
+     Statement capable  N YYYYYYYYY NNNNNNNNN YYYYYYYYY
+
+     Statement type     * SSSUUUIII SSSUUUIII SSSUUUIII
+
+     binlog_format      * SMRSMRSMR SMRSMRSMR SMRSMRSMR
+
+     Logged format      - SS-SS---- -RR-RR-RR SRRSRR-RR
+     Warning/Error      1 --2332444 5--5--6-- ---7--6--
+
+  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.
+
+  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)
+{
+  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()));
+
+  /*
+    We should not decide logging format if the binlog is closed or
+    binlogging is off, or if the statement is filtered out from the
+    binlog by filtering rules.
+  */
+  if (mysql_bin_log.is_open() && (options & OPTION_BIN_LOG) &&
+      !(variables.binlog_format == BINLOG_FORMAT_STMT &&
+        !binlog_filter->db_ok(db)))
+  {
+    /*
+      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=
+      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.
+    */
+    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::BINLOG_STMT_UNSAFE_SYSTEM_TABLE);
+      if (table->lock_type >= TL_WRITE_ALLOW_WRITE)
+      {
+        handler::Table_flags const flags= table->table->file->ha_table_flags();
+        DBUG_PRINT("info", ("table: %s; ha_table_flags: 0x%llx",
+                            table->table_name, flags));
+        if (prev_ht && prev_ht != table->table->file->ht)
+          multi_engine= TRUE;
+        prev_ht= table->table->file->ht;
+        flags_all_set &= flags;
+        flags_some_set |= flags;
+      }
+    }
+
+    DBUG_PRINT("info", ("flags_all_set: 0x%llx", flags_all_set));
+    DBUG_PRINT("info", ("flags_some_set: 0x%llx", flags_some_set));
+    DBUG_PRINT("info", ("multi_engine: %d", multi_engine));
+
+    int error= 0;
+    int unsafe_flags;
+
+    /*
+      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))
+    {
+      my_error((error= ER_BINLOG_MULTIPLE_ENGINES_AND_SELF_LOGGING_ENGINE),
+               MYF(0));
+    }
+
+    /* both statement-only and row-only engines involved */
+    if ((flags_all_set & (HA_BINLOG_STMT_CAPABLE | HA_BINLOG_ROW_CAPABLE)) == 0)
+    {
+      /*
+        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 ((unsafe_flags= lex->get_stmt_unsafe_flags()) != 0)
+      {
+        /*
+          3. Warning: Unsafe statement binlogged as statement since
+             storage engine is limited to statement-logging.
+        */
+        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! */
+    }
+    /* 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 ((unsafe_flags= lex->get_stmt_unsafe_flags()) != 0)
+        {
+          /*
+            7. Warning: Unsafe statement logged as statement due to
+               binlog_format = STATEMENT
+          */
+          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! */
+      }
+      /* 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_format_row_if_mixed();
+        }
+      }
+    }
+
+    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 "
+                        "and binlog_format = %d "
+                        "and binlog_filter->db_ok(db) = %d",
+                        mysql_bin_log.is_open(),
+                        (options & OPTION_BIN_LOG),
+                        variables.binlog_format,
+                        binlog_filter->db_ok(db)));
+#endif
+
+  DBUG_RETURN(0);
+}
+
+
 /*
   Implementation of interface to write rows to the binary log through the
   thread.  The thread is responsible for writing the rows it has
@@ -3550,7 +3856,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
@@ -3580,7 +3886,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);
@@ -3625,7 +3931,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
@@ -3707,8 +4013,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);
@@ -3720,28 +4024,119 @@ 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.
+/**
+  Auxiliary method used by @c binlog_query() to raise warnings.
 
-  This function should be called after the all calls to ha_*_row()
-  functions have been issued, but before tables are unlocked and
-  closed.
+  The type of warning and the type of unsafeness is stored in
+  THD::binlog_unsafe_warning_flags.
+*/
+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);
 
-  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.
+  /**
+    @note The order of the elements of this array must correspond to
+    the order of elements in enum_binlog_stmt_unsafe.
+  */
+  static const int explanations[LEX::BINLOG_STMT_UNSAFE_COUNT] =
+  {
+    ER_BINLOG_UNSAFE_LIMIT,
+    ER_BINLOG_UNSAFE_INSERT_DELAYED,
+    ER_BINLOG_UNSAFE_SYSTEM_TABLE,
+    ER_BINLOG_UNSAFE_TWO_AUTOINC_COLUMNS,
+    ER_BINLOG_UNSAFE_UDF,
+    ER_BINLOG_UNSAFE_SYSTEM_VARIABLE,
+    ER_BINLOG_UNSAFE_SYSTEM_FUNCTION
+  };
+  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;
+  }
 
-    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.
+  DBUG_PRINT("info", ("flags: 0x%x  err: %d", unsafe_type_flags, err));
 
-  RETURN VALUE
-    Error code, or 0 if no error.
+  /*
+    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,
+                          ER(ER_BINLOG_UNSAFE_WARNING_SHORT),
+                          ER(err), ER(explanations[unsafe_type]));
+      if (global_system_variables.log_warnings)
+        sql_print_warning(ER(ER_BINLOG_UNSAFE_WARNING_LONG),
+                          ER(err), ER(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
+  depending on the value of @c current_stmt_binlog_format_row field and
+  the value of the @c qtype parameter.
+
+  This function must be called:
+
+  - After the all calls to ha_*_row() functions have been issued.
+
+  - 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,
@@ -3766,52 +4161,46 @@ 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 && 
-      binlog_filter->db_ok(this->db))
-  {
-   /*
-     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 (global_system_variables.log_warnings &&
-        !(binlog_flags & BINLOG_FLAG_UNSAFE_STMT_PRINTED))
-    {
-      sql_print_warning("%s Statement: %.*s",
-                        ER(ER_BINLOG_UNSAFE_STATEMENT),
-                        MYSQL_ERRMSG_SIZE, query_arg);
-      binlog_flags|= BINLOG_FLAG_UNSAFE_STMT_PRINTED;
-    }
-  }
+  if (sql_log_bin_toplevel)
+    issue_unsafe_warnings();
 
   switch (qtype) {
+    /*
+      ROW_QUERY_TYPE means that the statement may be logged either in
+      row format or in statement format.  If
+      current_stmt_binlog_format is row, 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: %d",
+                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_format_row is row.
+
+      @todo Currently there are places that call this method with
+      STMT_QUERY_TYPE and current_stmt_binlog_format is row.  Fix those
+      places and add assert to ensure correct behavior. /Sven
     */
   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 f52d5fae76f..abba64bebed 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -1248,6 +1248,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.
 
@@ -1427,23 +1428,85 @@ public:
   int binlog_flush_pending_rows_event(bool stmt_end);
   int binlog_remove_pending_rows_event(bool clear_maps);
 
+  /**
+    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:
-  /*
-    Number of outstanding table maps, i.e., table maps in the
-    transaction cache.
+  /**
+    Indicates the format in which the current statement will be
+    logged.  This can only be set from @c decide_logging_format().
   */
-  uint binlog_table_maps;
+  enum_binlog_format current_stmt_binlog_format;
+
+  /**
+    Enumeration listing binlog-related warnings that a statement can
+    cause.
+  */
+  enum enum_binlog_stmt_warning {
 
-  enum enum_binlog_flag {
-    BINLOG_FLAG_UNSAFE_STMT_PRINTED,
-    BINLOG_FLAG_COUNT
+    /* ER_BINLOG_UNSAFE_AND_STMT_ENGINE affects current stmt */
+    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 with per-thread information regarding the status of the
-     binary log.
-   */
-  uint32 binlog_flags;
+    Bit field for the state of binlog warnings.
+
+    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_unsafe_warning_flags;
+
+  void issue_unsafe_warnings();
+
+  /*
+    Number of outstanding table maps, i.e., table maps in the
+    transaction cache.
+  */
+  uint binlog_table_maps;
 public:
   uint get_binlog_table_maps() const {
     return binlog_table_maps;
@@ -1755,8 +1818,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;
@@ -1908,21 +1969,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
   };
   
@@ -2126,31 +2178,51 @@ public:
   void set_n_backup_active_arena(Query_arena *set, Query_arena *backup);
   void restore_active_arena(Query_arena *set, Query_arena *backup);
 
-  inline void set_current_stmt_binlog_row_based_if_mixed()
+  /*
+    @todo Make these methods private or remove them completely.  Only
+    decide_logging_format should call them. /Sven
+  */
+  inline void set_current_stmt_binlog_format_row_if_mixed()
   {
+    DBUG_ENTER("set_current_stmt_binlog_format_row_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
       of executing a stored function/trigger (which is too late, see also
-      reset_current_stmt_binlog_row_based()); this condition will make their
+      reset_current_stmt_binlog_format_row()); this condition will make their
       tests fail and so force them to propagate the
       lex->binlog_row_based_if_mixed upwards to the caller.
     */
     if ((variables.binlog_format == BINLOG_FORMAT_MIXED) &&
         (in_sub_stmt == 0))
-      current_stmt_binlog_row_based= TRUE;
+      set_current_stmt_binlog_format_row();
+
+    DBUG_VOID_RETURN;
   }
-  inline void set_current_stmt_binlog_row_based()
+  inline void set_current_stmt_binlog_format_row()
   {
-    current_stmt_binlog_row_based= TRUE;
+    DBUG_ENTER("set_current_stmt_binlog_format_row");
+    current_stmt_binlog_format= BINLOG_FORMAT_ROW;
+    DBUG_VOID_RETURN;
   }
-  inline void clear_current_stmt_binlog_row_based()
+  inline void clear_current_stmt_binlog_format_row()
   {
-    current_stmt_binlog_row_based= FALSE;
+    DBUG_ENTER("clear_current_stmt_binlog_format_row");
+    current_stmt_binlog_format= BINLOG_FORMAT_STMT;
+    DBUG_VOID_RETURN;
   }
-  inline void reset_current_stmt_binlog_row_based()
+  inline void reset_current_stmt_binlog_format_row()
   {
+    DBUG_ENTER("reset_current_stmt_binlog_format_row");
     /*
       If there are temporary tables, don't reset back to
       statement-based. Indeed it could be that:
@@ -2165,19 +2237,19 @@ 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);
+      if (variables.binlog_format == BINLOG_FORMAT_ROW)
+        set_current_stmt_binlog_format_row();
+      else
+        clear_current_stmt_binlog_format_row();
     }
+    DBUG_VOID_RETURN;
   }
 
   /**
@@ -2281,7 +2353,11 @@ public:
     Protected with LOCK_thd_data mutex.
   */
   void set_query(char *query_arg, uint32 query_length_arg);
+
+  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 d2f90fa9288..7fa7e39e5bf 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 */
@@ -460,19 +460,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,
@@ -1031,15 +1018,16 @@ 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);
-  thd->clear_current_stmt_binlog_row_based();
+  thd->clear_current_stmt_binlog_format_row();
   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_format_row();
   DBUG_RETURN(error);
 }
 
@@ -1138,7 +1126,7 @@ end:
     {
       /*
         TRUNCATE must always be statement-based binlogged (not row-based) so
-        we don't test current_stmt_binlog_row_based.
+        we don't test current_stmt_binlog_format.
       */
       write_bin_log(thd, TRUE, thd->query, thd->query_length);
       my_ok(thd);		// This should return record count
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index a799cbea4c2..e5b03a1d44d 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1723,6 +1723,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;
@@ -1731,11 +1732,21 @@ 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_format_row_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();
+    thd.lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
+    thd.set_current_stmt_binlog_format_row_if_mixed();
 
     bzero((char*) &thd.net, sizeof(thd.net));		// Safety
     bzero((char*) &table_list, sizeof(table_list));	// Safety
@@ -1750,6 +1761,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()
   {
@@ -2329,12 +2341,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,
@@ -2780,7 +2786,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)))
@@ -2845,19 +2851,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
   */
@@ -3314,7 +3307,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);
@@ -3559,11 +3552,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)
       {
@@ -3587,7 +3580,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();
@@ -3606,7 +3599,7 @@ select_create::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
       push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
                           ER_TABLE_EXISTS_ERROR, ER(ER_TABLE_EXISTS_ERROR),
                           create_table->table_name);
-      if (thd->current_stmt_binlog_row_based)
+      if (thd->is_current_stmt_binlog_format_row())
         binlog_show_create_table(&(create_table->table), 1);
       table= create_table->table;
     }
@@ -3694,7 +3687,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];
@@ -3734,7 +3727,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 76fd5354c51..d3b719f1b72 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1043,25 +1043,144 @@ public:
     }
   }
 
+
   /**
-     Has the parser/scanner detected that this statement is unsafe?
-   */
+    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_SYSTEM_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);
+
+  /**
+    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.
+  */
   inline bool is_stmt_unsafe() const {
-    return binlog_stmt_flags & (1U << BINLOG_STMT_FLAG_UNSAFE);
+    return get_stmt_unsafe_flags() != 0;
   }
 
   /**
-     Flag the current (top-level) statement as unsafe.
+    Flag the current (top-level) statement as unsafe.
+    The flag will be reset after the statement has finished.
 
-     The flag will be reset after the statement has finished.
+    @param unsafe_type The type of unsafety: one of the @c
+    BINLOG_STMT_FLAG_UNSAFE_* flags in @c enum_binlog_stmt_flag.
+  */
+  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;
+  }
 
-   */
-  inline void set_stmt_unsafe() {
-    binlog_stmt_flags|= (1U << BINLOG_STMT_FLAG_UNSAFE);
+  /**
+    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_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;
   }
 
+  /**
+    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() {
-    binlog_stmt_flags&= ~(1U << BINLOG_STMT_FLAG_UNSAFE);
+    DBUG_ENTER("clear_stmt_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;
   }
 
   /**
@@ -1072,16 +1191,37 @@ public:
   { return sroutines_list.elements != 0; }
 
 private:
-  enum enum_binlog_stmt_flag {
-    BINLOG_STMT_FLAG_UNSAFE,
-    BINLOG_STMT_FLAG_COUNT
+
+  /**
+    Enumeration listing special types of statements.
+
+    Currently, the only possible type is ROW_INJECTION.
+  */
+  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_TYPE_ROW_INJECTION = 0,
+
+    /** The last element of this enumeration type. */
+    BINLOG_STMT_TYPE_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.
+  /**
+    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.
+
+    - The next BINLOG_STMT_TYPE_COUNT bits indicate if the statement
+      is of some special type.
+
+    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;
 };
@@ -1891,6 +2031,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 b7f33d51335..ab5f3165974 100644
--- a/sql/sql_load.cc
+++ b/sql/sql_load.cc
@@ -529,7 +529,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 ca27d476213..8dc3195044e 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5608,21 +5608,26 @@ 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();
+}
+
+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);
@@ -5666,15 +5671,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->reset_current_stmt_binlog_format_row();
+  thd->binlog_unsafe_warning_flags= 0;
 
   DBUG_PRINT("debug",
-             ("current_stmt_binlog_row_based: %d",
-              thd->current_stmt_binlog_row_based));
+             ("is_current_stmt_binlog_format_row(): %d",
+              thd->is_current_stmt_binlog_format_row()));
 
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 0ec8d91214c..5c18ff7690c 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1685,7 +1685,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 4ec104aedc3..84868cfb509 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1851,7 +1851,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)
@@ -1914,7 +1914,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       // removed temporary table
       tmp_table_deleted= 1;
       if (thd->variables.binlog_format == BINLOG_FORMAT_MIXED &&
-          thd->current_stmt_binlog_row_based)
+          thd->is_current_stmt_binlog_format_row())
       {
         if (built_tmp_query.is_empty()) 
         {
@@ -1948,7 +1948,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++;
       /*
@@ -2076,7 +2076,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))
       {
         /*
@@ -2088,7 +2088,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() &&
                tmp_table_deleted)
       {
         if (non_temp_tables_count > 0)
@@ -3549,8 +3549,8 @@ static inline void write_create_table_bin_log(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);
 }
@@ -5372,7 +5372,7 @@ binlog:
   /*
     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
@@ -7257,7 +7257,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;
   }
@@ -7419,7 +7419,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..bd5ce822ff1 100644
--- a/sql/sql_udf.cc
+++ b/sql/sql_udf.cc
@@ -437,8 +437,8 @@ 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)
-    thd->clear_current_stmt_binlog_row_based();
+  if (thd->is_current_stmt_binlog_format_row())
+    thd->clear_current_stmt_binlog_format_row();
 
   rw_wrlock(&THR_LOCK_udf);
   if ((hash_search(&udf_hash,(uchar*) udf->name.str, udf->name.length)))
@@ -540,8 +540,8 @@ 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)
-    thd->clear_current_stmt_binlog_row_based();
+  if (thd->is_current_stmt_binlog_format_row())
+    thd->clear_current_stmt_binlog_format_row();
 
   rw_wrlock(&THR_LOCK_udf);  
   if (!(udf=(udf_func*) hash_search(&udf_hash,(uchar*) udf_name->str,
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 152613c0009..b0085864108 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -878,19 +878,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_view.cc b/sql/sql_view.cc
index 43d0b9fade0..8b00b8f0d9b 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());
 
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index a18f57bf9cf..1a19be81a44 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -7225,7 +7225,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_SYSTEM_FUNCTION);
             Lex->safe_to_cache_query= 0;
           }
         | DATE_SYM '(' expr ')'
@@ -7380,7 +7380,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_SYSTEM_FUNCTION);
             Lex->safe_to_cache_query=0;
           }
         | YEAR_SYM '(' expr ')'
@@ -8110,7 +8110,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);
           }
         ;
 
@@ -8956,7 +8956,10 @@ opt_limit_clause:
         ;
 
 limit_clause:
-          LIMIT limit_options {}
+          LIMIT limit_options
+          {
+            Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
+          }
         ;
 
 limit_options:
@@ -9018,6 +9021,7 @@ delete_limit_clause:
           {
             SELECT_LEX *sel= Select;
             sel->select_limit= $2;
+            Lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_LIMIT);
             sel->explicit_limit= 1;
           }
         ;
@@ -9469,13 +9473,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(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
+        }
         | 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(LEX::BINLOG_STMT_UNSAFE_INSERT_DELAYED);
+        }
         ;
 
 insert2: