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

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

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

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

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

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


mysql-test/extra/rpl_tests/create_recursive_construct.inc:
  Added auxiliary file used by binlog_unsafe.test to create and
  execute recursive constructs
  (functions/procedures/triggers/views/prepared statements).
mysql-test/extra/rpl_tests/rpl_foreign_key.test:
  removed unnecessary set @@session.binlog_format
mysql-test/extra/rpl_tests/rpl_insert_delayed.test:
  Filter out table id from table map events in binlog listing.
  Got rid of $binlog_format_statement.
mysql-test/extra/rpl_tests/rpl_ndb_apply_status.test:
  disable warnings around call to unsafe procedure
mysql-test/include/rpl_udf.inc:
  Disabled warnings for code that generates warnings
  for some binlog formats. That would otherwise cause
  inconsistencies in the result file.
mysql-test/r/mysqldump.result:
  Views are now unsafe if they contain a LIMIT clause.
  That fixed BUG#45831. Due to BUG#45832, a warning is
  printed for the CREATE VIEW statement.
mysql-test/r/sp_trans.result:
  Unsafe statements in stored procedures did not give a warning if
  binlog_format=statement. This is BUG#45824. Now they do, so this
  result file gets a new warning.
mysql-test/suite/binlog/r/binlog_multi_engine.result:
  Error message changed.
mysql-test/suite/binlog/r/binlog_statement_insert_delayed.result:
  INSERT DELAYED didn't generate a warning when binlog_format=STATEMENT.
  That was BUG#45825. Now there is a warning, so result file needs to be
  updated.
mysql-test/suite/binlog/r/binlog_stm_ps.result:
  Changed error message.
mysql-test/suite/binlog/r/binlog_unsafe.result:
  updated result file:
   - error message changed
   - added test for most combinations of unsafe constructs invoked
     from recursive constructs
   - INSERT DELAYED now gives a warning (because BUG#45826 is fixed)
   - INSERT..SELECT..LIMIT now gives a warning from inside recursive
     constructs (because BUG#45785 was fixed)
   - When a recursive construct (e.g., stored proc or function)
     contains more than one statement, at least one of which is
     unsafe, then all statements in the recursive construct give
     warnings. This is a new bug introduced by this changeset.
     It will be addressed in a post-push fix.
mysql-test/suite/binlog/t/binlog_innodb.test:
  Changed error code for innodb updates with READ COMMITTED or 
  READ UNCOMMITTED transaction isolation level and
  binlog_format=statement.
mysql-test/suite/binlog/t/binlog_multi_engine.test:
  The error code has changed for statements where more than one
  engine is involved and one of them is self-logging.
mysql-test/suite/binlog/t/binlog_unsafe-master.opt:
  Since binlog_unsafe now tests unsafe-ness of UDF's, we need an extra
  flag in the .opt file.
mysql-test/suite/binlog/t/binlog_unsafe.test:
   - Clarified comment.
   - Rewrote first part of test. Now it tests not only unsafe variables
     and functions, but also unsafe-ness due to INSERT..SELECT..LIMIT,
     INSERT DELAYED, insert into two autoinc columns, use of UDF's, and
     access to log tables in the mysql database.
     Also, in addition to functions, procedures, triggers, and prepared
     statements, it now also tests views; and it constructs recursive
     calls in two levels by combining these recursive constructs.
     Part of the logic is in extra/rpl_tests/create_recursive_construct.inc.
   - added tests for all special system variables that should not be unsafe.
   - added specific tests for BUG#45785 and BUG#45825
mysql-test/suite/rpl/r/rpl_events.result:
  updated result file
mysql-test/suite/rpl/r/rpl_extraColmaster_innodb.result:
  updated result file
mysql-test/suite/rpl/r/rpl_extraColmaster_myisam.result:
  updated result file
mysql-test/suite/rpl/r/rpl_foreign_key_innodb.result:
  updated result file
mysql-test/suite/rpl/r/rpl_idempotency.result:
  updated result file
mysql-test/suite/rpl/r/rpl_mix_found_rows.result:
  Split rpl_found_rows.test into rpl_mix_found_rows.test (a new file) and
  rpl_stm_found_rows.test (renamed rpl_found_rows.test). This file equals
  the second half of the old rpl_found_rows.result, with the following
  modifications:
   - minor formatting changes
   - additional initialization
mysql-test/suite/rpl/r/rpl_mix_insert_delayed.result:
  Moved out code operating in mixed mode from rpl_stm_insert_delayed
  (into rpl_mix_insert_delayed) and got rid of explicit setting of
  binlog format.
mysql-test/suite/rpl/r/rpl_rbr_to_sbr.result:
  updated result file
mysql-test/suite/rpl/r/rpl_row_idempotency.result:
  Moved the second half of rpl_idempotency.test, which only
  executed in row mode, to rpl_row_idempotency.test. This is
  the new result file.
mysql-test/suite/rpl/r/rpl_row_insert_delayed.result:
  Got rid of unnecessary explicit setting of binlog format.
mysql-test/suite/rpl/r/rpl_stm_found_rows.result:
  Split rpl_found_rows.test into rpl_mix_found_rows.test (a new file) and
  rpl_stm_found_rows.test (renamed rpl_found_rows.test). Changes in
  this file:
   - minor formatting changes
   - warning is now issued for unsafe statements inside procedures
     (since BUG#45824 is fixed)
   - second half of file is moved to rpl_mix_found_rows.result
mysql-test/suite/rpl/r/rpl_stm_insert_delayed.result:
  Moved out code operating in mixed mode from rpl_stm_insert_delayed
  (into rpl_mix_insert_delayed) and got rid of explicit setting of
  binlog format.
mysql-test/suite/rpl/r/rpl_stm_loadfile.result:
  error message changed
mysql-test/suite/rpl/r/rpl_temporary_errors.result:
  updated result file
mysql-test/suite/rpl/r/rpl_udf.result:
  Remove explicit set of binlog format (and triplicate test execution)
  and rely on test system executing the test in all binlog formats.
mysql-test/suite/rpl/t/rpl_bug31076.test:
  Test is only valid in mixed or row mode since it generates row events.
mysql-test/suite/rpl/t/rpl_events.test:
  Removed explicit set of binlog_format and removed duplicate testing.
  Instead, we rely on the test system to try all binlog formats.
mysql-test/suite/rpl/t/rpl_extraColmaster_innodb.test:
  Removed triplicate testing and instead relying on test system.
  Test is only relevant for row format since statement-based replication
  cannot handle extra columns on master.
mysql-test/suite/rpl/t/rpl_extraColmaster_myisam.test:
  Removed triplicate testing and instead relying on test system.
  Test is only relevant for row format since statement-based replication
  cannot handle extra columns on master.
mysql-test/suite/rpl/t/rpl_idempotency-slave.opt:
  Removed .opt file to avoid server restarts.
mysql-test/suite/rpl/t/rpl_idempotency.test:
  - Moved out row-only tests to a new test file, rpl_row_idempotency.test.
    rpl_idempotency now only contains tests that execute in all
    binlog_formats.
  - While I was here, also removed .opt file to avoid server restarts.
    The slave_exec_mode is now set inside the test instead.
mysql-test/suite/rpl/t/rpl_mix_found_rows.test:
  Split rpl_found_rows.test into rpl_mix_found_rows.test (a new file) and
  rpl_stm_found_rows.test (renamed rpl_found_rows.test). This file
  contains the second half of the original rpl_found_rows.test with the
  follwing changes:
   - initialization
   - removed SET_BINLOG_FORMAT and added have_binlog_format_mixed.inc
   - minor formatting changes
mysql-test/suite/rpl/t/rpl_mix_insert_delayed.test:
  Moved out code operating in mixed mode from rpl_stm_insert_delayed
  (into rpl_mix_insert_delayed) and got rid of explicit setting of
  binlog format.
mysql-test/suite/rpl/t/rpl_rbr_to_sbr.test:
  Test cannot execute in statement mode, since we no longer
  switch to row format when binlog_format=statement.
  Enforced mixed mode throughout the test.
mysql-test/suite/rpl/t/rpl_row_idempotency.test:
  Moved the second half of rpl_idempotency.test, which only
  executed in row mode, to this new file. We now rely on the
  test system to set binlog format.
mysql-test/suite/rpl/t/rpl_row_insert_delayed.test:
   - Got rid of unnecessary explicit setting of binlog format.
   - extra/rpl_tests/rpl_insert_delayed.test does not need the
     $binlog_format_statement variable any more, so that was
     removed.
mysql-test/suite/rpl/t/rpl_slave_skip.test:
  The test switches binlog_format internally and master generates both
  row and statement events. Hence, the slave must be able to log in both
  statement and row format. Hence test was changed to only execute in
  mixed mode.
mysql-test/suite/rpl/t/rpl_stm_found_rows.test:
  Split rpl_found_rows.test into rpl_mix_found_rows.test (a new file) and
  rpl_stm_found_rows.test (renamed rpl_found_rows.test). Changes in
  this file:
   - minor formatting changes
   - added have_binlog_format_statement and removed SET BINLOG_FORMAT.
   - second half of file is moved to rpl_mix_found_rows.test
   - added cleanup code
mysql-test/suite/rpl/t/rpl_stm_insert_delayed.test:
  Moved out code operating in mixed mode from rpl_stm_insert_delayed
  (into rpl_mix_insert_delayed) and got rid of explicit setting of
  binlog format.
mysql-test/suite/rpl/t/rpl_switch_stm_row_mixed.test:
  The test switches binlog_format internally and master generates both
  row and statement events. Hence, the slave must be able to log in both
  statement and row format. Hence test was changed to only execute in
  mixed mode on slave.
mysql-test/suite/rpl/t/rpl_temporary_errors.test:
  Removed explicit set of binlog format. Instead, the test now only
  executes in row mode.
mysql-test/suite/rpl/t/rpl_udf.test:
  Remove explicit set of binlog format (and triplicate test execution)
  and rely on test system executing the test in all binlog formats.
mysql-test/suite/rpl_ndb/combinations:
  Added combinations file for rpl_ndb.
mysql-test/suite/rpl_ndb/r/rpl_ndb_binlog_format_errors.result:
  new result file
mysql-test/suite/rpl_ndb/r/rpl_ndb_circular_simplex.result:
  updated result file
mysql-test/suite/rpl_ndb/t/rpl_ndb_2innodb.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_2myisam.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_basic.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_binlog_format_errors-master.opt:
  new option file
mysql-test/suite/rpl_ndb/t/rpl_ndb_binlog_format_errors-slave.opt:
  new option file
mysql-test/suite/rpl_ndb/t/rpl_ndb_binlog_format_errors.test:
  New test case to verify all errors and warnings generated by
  decide_logging_format.
mysql-test/suite/rpl_ndb/t/rpl_ndb_blob.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_blob2.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_circular.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_circular_simplex.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
  While I was here, also made the test clean up after itself.
mysql-test/suite/rpl_ndb/t/rpl_ndb_commit_afterflush.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_ctype_ucs2_def.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_delete_nowhere.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_do_db.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_do_table.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_func003.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_innodb_trans.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_insert_ignore.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_mixed_engines_transactions.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_multi_update3.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_rep_ignore.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_row_001.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_sp003.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_sp006.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/suite/rpl_ndb/t/rpl_ndb_trig004.test:
  The test needs slave to be able to switch to row mode, so the
  test was changed to only execute in mixed and row mode.
mysql-test/t/partition_innodb_stmt.test:
  Changed error code for innodb updates with READ COMMITTED or 
  READ UNCOMMITTED transaction isolation level and
  binlog_format=statement.
sql/event_db_repository.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/events.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/ha_ndbcluster_binlog.cc:
  reset_current_stmt_binlog_row_based() is not a no-op for the ndb_binlog
  thread any more. Instead, the ndb_binlog thread now forces row mode both
  initially and just after calling mysql_parse.  (mysql_parse() is the only
  place where reset_current_stmt_binlog_row_based() may be called from
  the ndb_binlog thread, so these are the only two places that need to
  change.)
sql/ha_partition.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/handler.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/item_create.cc:
  Added DBUG_ENTER to some functions, to be able to trace when
  set_stmt_unsafe is called.
sql/log.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/log_event.cc:
   - Moved logic for changing to row format out of do_apply_event (and into
     decide_logging_format).
   - Added @todo comment for post-push cleanup.
sql/log_event_old.cc:
  Move logic for changing to row format out of do_apply_event (and into
  decide_logging_format).
sql/mysql_priv.h:
  Make decide_logging_format() a member of the THD class, for two reasons:
   - It is natural from an object-oriented perspective.
   - decide_logging_format() needs to access private members of THD
     (specifically, the new binlog_warning_flags field).
sql/rpl_injector.cc:
  Removed call to set_current_stmt_binlog_row_based().
  From now on, only decide_logging_fromat is allowed to modify
  current_stmt_binlog_row_based. This call is from the ndb_binlog
  thread, mostly executing code in ha_ndbcluster_binlog.cc.
  This call can be safely removed, because:
   - current_stmt_binlog_row_based is initialized for the ndb_binlog
     thread's THD object when the THD object is created. So we're
     not going to read uninitialized memory.
   - The behavior of ndb_binlog thread does not use the state of the
     current_stmt_binlog_row_based. It is conceivable that the
     ndb_binlog thread would rely on the current_stmt_binlog_format
     in two situations:
      (1) when it calls mysql_parse;
      (2) when it calls THD::binlog_query.
     In case (1), it always clears THD::options&OPTION_BIN_LOG (because
     run_query() in ha_ndbcluster_binlog.cc is only called with
     disable_binlogging = TRUE).
     In case (2), it always uses qtype=STMT_QUERY_TYPE.
sql/set_var.cc:
  Added @todo comment for post-push cleanup.
sql/share/errmsg.txt:
  Added new error messages and clarified ER_BINLOG_UNSAFE_STATEMENT.
sql/sp.cc:
  Added DBUG_ENTER, to be able to trace when set_stmt_unsafe is called.
  Got rid of MYSQL_QUERY_TYPE: it was equivalent to STMT_QUERY_TYPE.
sql/sp_head.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/sp_head.h:
  Added DBUG_ENTER, to be able to trace when set_stmt_unsafe is called.
sql/sql_acl.cc:
  Got rid of MYSQL_QUERY_TYPE: it was equivalent to STMT_QUERY_TYPE.
sql/sql_base.cc:
   - Made decide_logging_format take care of all logic for deciding the
     logging format, and for determining the related warnings and errors.
     See comment above decide_logging_format for details.
   - Made decide_logging_format a member function of THD, since it needs
     to access private members of THD and since its purpose is to update
     the state of a THD object.
   - Added DBUG_ENTER, to be able to trace when set_stmt_unsafe is called.
sql/sql_class.cc:
  - Moved logic for determining unsafe warnings away from THD::binlog_query
    (and into decide_logging_format()). Now, it works like this:
    1. decide_logging_format detects that the current statement shall
       produce a warning, if it ever makes it to the binlog
    2. decide_logging_format sets a flag of THD::binlog_warning_flags.
    3. THD::binlog_query reads the flag. If the flag is set, it generates
       a warning.
  - Use member function to read current_stmt_binlog_row_based.
sql/sql_class.h:
  - Added THD::binlog_warning_flags (see sql_class.cc for explanation).
  - Made decide_logging_format() and reset_for_next_command() member
    functions of THD (instead of standalone functions). This was needed
    for two reasons: (1) the functions need to access the private member
    THD::binlog_warning_flags; (2) the purpose of these functions is to
    update the staet of a THD object, so from an object-oriented point
    of view they should be member functions.
  - Encapsulated current_stmt_binlog_row_based, so it is now private and
    can only be accessed from a member function. Also changed the
    data type to an enumeration instead of a bool.
  - Removed MYSQL_QUERY_TYPE, because it was equivalent to
    STMT_QUERY_TYPE anyways.
  - When reset_current_stmt_binlog_row_based was called from the
    ndb_binlog thread, it would behave as a no-op. This special
    case has been removed, and the behavior of
    reset_current_stmt_binlog_row_based does not depend on which thread
    calls it any more. The special case did not serve any purpose,
    since the ndb binlog thread did not take the
    current_stmt_binlog_row_based flag into account anyways.
sql/sql_delete.cc:
  - Moved logic for setting row format for DELETE..LIMIT away from
    mysql_prepare_delete.
    (Instead, we mark the statement as unsafe at parse time (sql_yacc.yy)
    and rely on decide_logging_format() (sql_class.cc) to set row format.)
    This is part of the fix for BUG#45831.
  - Use member function to read current_stmt_binlog_row_based.
sql/sql_insert.cc:
   - Removed unnecessary calls to thd->lex->set_stmt_unsafe() and
     thd->set_current_stmt_binlog_row_based_if_mixed() from
     handle_delayed_insert(). The calls are unnecessary because they
     have already been made; they were made in the constructor of
     the `di' object.
   - Since decide_logging_format() is now a member function of THD, code
     that calls decide_logging_format() had to be updated.
   - Added DBUG_ENTER call, to be able to trace when set_stmt_unsafe is
     called.
   - Moved call to set_stmt_unsafe() for INSERT..SELECT..LIMIT away from
     mysql_insert_select_prepare() (and into decide_logging_format).
     This is part of the fix for BUG#45831.
   - Use member function to read current_stmt_binlog_row_based.
sql/sql_lex.h:
   - Added the flag BINLOG_STMT_FLAG_ROW_INJECTION to enum_binlog_stmt_flag.
     This was necessary so that a statement can identify itself as a row
     injection.
   - Added appropriate setter and getter functions for the new flag.
   - Added or clarified some comments.
   - Added DBUG_ENTER()
sql/sql_load.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/sql_parse.cc:
   - Made mysql_reset_thd_for_next_command() clear thd->binlog_warning_flags.
   - Since thd->binlog_warning_flags is private, it must be set in a
     member function of THD. Hence, moved the body of
     mysql_reset_thd_for_next_command() to the new member function
     THD::reset_thd_for_next_command(), and made
     mysql_reset_thd_for_next_command() call
     THD::reset_thd_for_next_command().
   - Removed confusing comment.
   - Use member function to read current_stmt_binlog_row_based.
sql/sql_repl.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/sql_table.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/sql_udf.cc:
  Use member function to read current_stmt_binlog_row_based.
sql/sql_update.cc:
  Moved logic for setting row format for UPDATE..LIMIT away from
  mysql_prepare_update.
  (Instead, we mark the statement as unsafe at parse time (sql_yacc.yy)
  and rely on decide_logging_format() (sql_class.cc) to set row format.)
  This is part of the fix for BUG#45831.
sql/sql_yacc.yy:
  Made INSERT DELAYED, INSERT..SELECT..LIMIT, UPDATE..LIMIT, and
  DELETE..LIMIT mark themselves as unsafe at parse time (instead
  of at execution time).
  This is part of the fixes BUG#45831 and BUG#45825.
storage/example/ha_example.cc:
  Made exampledb accept inserts. This was needed by the new test case
  rpl_ndb_binlog_format_errors, because it needs an engine that
  is statement-only (and accepts inserts).
storage/example/ha_example.h:
  Made exampledb a statement-only engine instead of a row-only engine.
  No existing test relied exampledb's row-only capabilities. The new
  test case rpl_ndb_binlog_format_errors needs an engine that is
  statement-only.
storage/innobase/handler/ha_innodb.cc:
  - Changed error error code and message given by innodb when 
    binlog_format=STATEMENT and transaction isolation level is
    READ COMMITTED or READ UNCOMMITTED.
  - While I was here, also simplified the condition for
    checking when to give the error.

30 files changed, 525 insertions, 295 deletions

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