BUG#53452 Inconsistent behavior of binlog_direct_non_transactional_updates with

temp table
            
This patch introduces two key changes in the replication's behavior.
            
Firstly, it reverts part of BUG#51894 which puts any update to temporary tables
into the trx-cache. Now, updates to temporary tables are handled according to
the type of their engines as a regular table.
            
Secondly, an unsafe mixed statement, (i.e. a statement that access transactional
table as well non-transactional or temporary table, and writes to any of them),
are written into the trx-cache in order to minimize errors in the execution when
the statement logging format is in use.
            
Such changes has a direct impact on which statements are classified as unsafe
statements and thus part of BUG#53259 is reverted.

11 files changed, 569 insertions, 331 deletions

diff --git a/sql/log.cc b/sql/log.cc
index 86f8100be07..5ca22d154b1 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -4272,7 +4272,7 @@ bool use_trans_cache(const THD* thd, bool is_transactional)
     (binlog_cache_mngr*) thd_get_ha_data(thd, binlog_hton);
 
   return
-    ((thd->variables.binlog_format != BINLOG_FORMAT_STMT ||
+    ((thd->is_current_stmt_binlog_format_row() ||
      thd->variables.binlog_direct_non_trans_update) ? is_transactional :
      (is_transactional || !cache_mngr->trx_cache.empty()));
 }
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d224f555cf2..c1f7836e08a 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -678,11 +678,11 @@ Log_event::Log_event(THD* thd_arg, uint16 flags_arg, bool using_trans)
 {
   server_id=	thd->server_id;
   when=		thd->start_time;
-  cache_type= ((using_trans || stmt_has_updated_trans_table(thd) ||
-               (thd->lex->stmt_accessed_temp_table() &&
-               trans_has_updated_trans_table(thd)))
-               ? Log_event::EVENT_TRANSACTIONAL_CACHE :
-               Log_event::EVENT_STMT_CACHE);
+
+  if (using_trans)
+    cache_type= Log_event::EVENT_TRANSACTIONAL_CACHE;
+  else
+    cache_type= Log_event::EVENT_STMT_CACHE;
 }
 
 /**
@@ -2520,21 +2520,18 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
 
   LEX *lex= thd->lex;
   /*
-    TRUE defines that either a trx-cache or stmt-cache must be used
-    and wrapped by a BEGIN...COMMIT. Otherwise, the statement will
-    be written directly to the binary log without being wrapped by
-    a BEGIN...COMMIT.
+    Defines that the statement will be written directly to the binary log
+    without being wrapped by a BEGIN...COMMIT. Otherwise, the statement
+    will be written to either the trx-cache or stmt-cache.
 
-    Note that a cache will not be used if the parameter direct is
-    TRUE.
+    Note that a cache will not be used if the parameter direct is TRUE.
   */
   bool use_cache= FALSE;
   /*
-    TRUE defines that the trx-cache must be used and by consequence
-    the use_cache is TRUE.
+    TRUE defines that the trx-cache must be used and by consequence the
+    use_cache is TRUE.
 
-    Note that a cache will not be used if the parameter direct is
-    TRUE.
+    Note that a cache will not be used if the parameter direct is TRUE.
   */
   bool trx_cache= FALSE;
   cache_type= Log_event::EVENT_INVALID_CACHE;
@@ -2542,16 +2539,14 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   switch (lex->sql_command)
   {
     case SQLCOM_DROP_TABLE:
-      use_cache= trx_cache= (lex->drop_temporary &&
-                            thd->in_multi_stmt_transaction_mode());
+      use_cache= (lex->drop_temporary && thd->in_multi_stmt_transaction_mode());
     break;
 
     case SQLCOM_CREATE_TABLE:
-      use_cache= trx_cache=
-                 ((lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
-                   thd->in_multi_stmt_transaction_mode()) ||
-                 (lex->select_lex.item_list.elements &&
+      trx_cache= (lex->select_lex.item_list.elements &&
                   thd->is_current_stmt_binlog_format_row());
+      use_cache= ((lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
+                   thd->in_multi_stmt_transaction_mode()) || trx_cache;
       break;
     case SQLCOM_SET_OPTION:
       use_cache= trx_cache= (lex->autocommit ? FALSE : TRUE);
@@ -2570,14 +2565,14 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   {
     cache_type= Log_event::EVENT_NO_CACHE;
   }
+  else if (using_trans || trx_cache || stmt_has_updated_trans_table(thd) ||
+           thd->lex->is_mixed_stmt_unsafe(thd->in_multi_stmt_transaction_mode(),
+                                          thd->variables.binlog_direct_non_trans_update,
+                                          trans_has_updated_trans_table(thd),
+                                          thd->tx_isolation))
+    cache_type= Log_event::EVENT_TRANSACTIONAL_CACHE;
   else
-  {
-    cache_type= ((using_trans || stmt_has_updated_trans_table(thd) || trx_cache ||
-                 (thd->lex->stmt_accessed_temp_table() &&
-                 trans_has_updated_trans_table(thd)))
-                 ? Log_event::EVENT_TRANSACTIONAL_CACHE :
-                 Log_event::EVENT_STMT_CACHE);
-  }
+    cache_type= Log_event::EVENT_STMT_CACHE;
   DBUG_ASSERT(cache_type != Log_event::EVENT_INVALID_CACHE);
   DBUG_PRINT("info",("Query_log_event has flags2: %lu  sql_mode: %lu",
                      (ulong) flags2, sql_mode));
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index eaa3e27cf29..a0eca1b78ff 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -4513,6 +4513,8 @@ int mysqld_main(int argc, char **argv)
   init_status_vars();
   if (opt_bootstrap) /* If running with bootstrap, do not start replication. */
     opt_skip_slave_start= 1;
+
+  binlog_unsafe_map_init();
   /*
     init_slave() must be called after the thread keys are created.
     Some parts of the code (e.g. SHOW STATUS LIKE 'slave_running' and other
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 243a551d24b..8be29eb12f5 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -2042,13 +2042,17 @@ TABLE *find_temporary_table(THD *thd, TABLE_LIST *table_list)
   thd->temporary_tables list, it's impossible to tell here whether
   we're dealing with an internal or a user temporary table.
 
+  If is_trans is not null, we return the type of the table:
+  either transactional (e.g. innodb) as TRUE or non-transactional
+  (e.g. myisam) as FALSE.
+
   @retval  0  the table was found and dropped successfully.
   @retval  1  the table was not found in the list of temporary tables
               of this thread
   @retval -1  the table is in use by a outer query
 */
 
-int drop_temporary_table(THD *thd, TABLE_LIST *table_list)
+int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans)
 {
   TABLE *table;
   DBUG_ENTER("drop_temporary_table");
@@ -2065,6 +2069,9 @@ int drop_temporary_table(THD *thd, TABLE_LIST *table_list)
     DBUG_RETURN(-1);
   }
 
+  if (is_trans != NULL)
+    *is_trans= table->file->has_transactions();
+
   /*
     If LOCK TABLES list is not empty and contains this table,
     unlock the table and remove the table from this list.
diff --git a/sql/sql_base.h b/sql/sql_base.h
index 2b08d8fa40e..e83c3cf8097 100644
--- a/sql/sql_base.h
+++ b/sql/sql_base.h
@@ -220,7 +220,7 @@ bool close_thread_table(THD *thd, TABLE **table_ptr);
 bool close_temporary_tables(THD *thd);
 TABLE_LIST *unique_table(THD *thd, TABLE_LIST *table, TABLE_LIST *table_list,
                          bool check_alias);
-int drop_temporary_table(THD *thd, TABLE_LIST *table_list);
+int drop_temporary_table(THD *thd, TABLE_LIST *table_list, bool *is_trans);
 void close_temporary_table(THD *thd, TABLE *table, bool free_share,
                            bool delete_table);
 void close_temporary(TABLE *table, bool free_share, bool delete_table);
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index eff7d8c1d90..bde56777d52 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -3775,124 +3775,16 @@ int THD::decide_logging_format(TABLE_LIST *tables)
     int error= 0;
     int unsafe_flags;
 
-    /*
-      Classify a statement as unsafe when there is a mixed statement and an
-      on-going transaction at any point of the execution if:
-
-        1. The mixed statement is about to update a transactional table and
-        a non-transactional table.
-
-        2. The mixed statement is about to update a temporary transactional
-        table and a non-transactional table.
-      
-        3. The mixed statement is about to update a transactional table and
-        read from a non-transactional table.
-
-        4. The mixed statement is about to update a temporary transactional
-        table and read from a non-transactional table.
-
-        5. The mixed statement is about to update a non-transactional table
-        and read from a transactional table when the isolation level is
-        lower than repeatable read.
-
-      After updating a transactional table if:
-
-        6. The mixed statement is about to update a non-transactional table
-        and read from a temporary transactional table.
- 
-        7. The mixed statement is about to update a non-transactional table
-        and read from a temporary transactional table.
-
-        8. The mixed statement is about to update a non-transactionala table
-        and read from a temporary non-transactional table.
-     
-        9. The mixed statement is about to update a temporary non-transactional
-        table and update a non-transactional table.
-     
-        10. The mixed statement is about to update a temporary non-transactional
-        table and read from a non-transactional table.
-     
-        11. A statement is about to update a non-transactional table and the
-        option variables.binlog_direct_non_trans_update is OFF.
-
-      The reason for this is that locks acquired may not protected a concurrent
-      transaction of interfering in the current execution and by consequence in
-      the result. In particular, if there is an on-going transaction and a
-      transactional table was already updated, a temporary table must be written
-      to the binary log in the boundaries of the on-going transaction and as
-      such we artificially classify them as transactional.
-    */
-    if (in_multi_stmt_transaction_mode())
-    {
-      my_bool mixed_unsafe= FALSE;
-      my_bool non_trans_unsafe= FALSE;
-
-      /* Case 1. */
-      if (lex->stmt_accessed_table(LEX::STMT_WRITES_TRANS_TABLE) &&
-          lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE))
-        mixed_unsafe= TRUE;
-      /* Case 2. */
-      else if (lex->stmt_accessed_table(LEX::STMT_WRITES_TEMP_TRANS_TABLE) &&
-               lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE))
-        mixed_unsafe= TRUE;
-      /* Case 3. */
-      else if (lex->stmt_accessed_table(LEX::STMT_WRITES_TRANS_TABLE) &&
-               lex->stmt_accessed_table(LEX::STMT_READS_NON_TRANS_TABLE))
-        mixed_unsafe= TRUE;
-      /* Case 4. */
-      else if (lex->stmt_accessed_table(LEX::STMT_WRITES_TEMP_TRANS_TABLE) &&
-               lex->stmt_accessed_table(LEX::STMT_READS_NON_TRANS_TABLE))
-        mixed_unsafe= TRUE;
-      /* Case 5. */
-      else if (lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE) &&
-               lex->stmt_accessed_table(LEX::STMT_READS_TRANS_TABLE) &&
-               tx_isolation < ISO_REPEATABLE_READ)
-        /*
-           By default, InnoDB operates in REPEATABLE READ and with the option
-           --innodb-locks-unsafe-for-binlog disabled. In this case, InnoDB uses
-           next-key locks for searches and index scans, which prevents phantom
-           rows.
- 
-           This is scenario is safe for Innodb. However, there are no means to
-           transparently get this information. Therefore, we need to improve this
-           and change the storage engines to report somehow when an execution is
-           safe under an isolation level & binary logging format.
-        */
-        mixed_unsafe= TRUE;
-
-      if (trans_has_updated_trans_table(this))
-      {
-        /* Case 6. */
-        if (lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE) &&
-            lex->stmt_accessed_table(LEX::STMT_READS_TRANS_TABLE))
-          mixed_unsafe= TRUE;
-        /* Case 7. */
-        else if (lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE) &&
-                 lex->stmt_accessed_table(LEX::STMT_READS_TEMP_TRANS_TABLE))
-          mixed_unsafe= TRUE;
-        /* Case 8. */
-        else if (lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE) &&
-                 lex->stmt_accessed_table(LEX::STMT_READS_TEMP_NON_TRANS_TABLE))
-          mixed_unsafe= TRUE;
-        /* Case 9. */
-        else if (lex->stmt_accessed_table(LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE) &&
-                 lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE))
-          mixed_unsafe= TRUE;
-        /* Case 10. */
-        else if (lex->stmt_accessed_table(LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE) &&
-                 lex->stmt_accessed_table(LEX::STMT_READS_NON_TRANS_TABLE))
-        mixed_unsafe= TRUE;
-        /* Case 11. */
-        else if (!variables.binlog_direct_non_trans_update &&
-                 lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE))
-          non_trans_unsafe= TRUE;
-      }
-
-      if (mixed_unsafe)
-        lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_MIXED_STATEMENT);
-      else if (non_trans_unsafe)
-        lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_NONTRANS_AFTER_TRANS);
-    }
+    bool multi_stmt_trans= in_multi_stmt_transaction_mode();
+    bool trans_table= trans_has_updated_trans_table(this);
+    bool binlog_direct= variables.binlog_direct_non_trans_update;
+
+    if (lex->is_mixed_stmt_unsafe(multi_stmt_trans, binlog_direct,
+                                  trans_table, tx_isolation))
+      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_MIXED_STATEMENT);
+    else if (multi_stmt_trans && trans_table && !binlog_direct &&
+             lex->stmt_accessed_table(LEX::STMT_WRITES_NON_TRANS_TABLE))
+      lex->set_stmt_unsafe(LEX::BINLOG_STMT_UNSAFE_NONTRANS_AFTER_TRANS);
 
     /*
       If more than one engine is involved in the statement and at
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 2314dbbeaee..79a5cfdbbe2 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3579,7 +3579,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
     if (!mysql_create_table_no_lock(thd, create_table->db,
                                     create_table->table_name,
                                     create_info, alter_info, 0,
-                                    select_field_count))
+                                    select_field_count, NULL))
     {
       DBUG_EXECUTE_IF("sleep_create_select_before_open", my_sleep(6000000););
 
@@ -3611,7 +3611,7 @@ static TABLE *create_table_from_items(THD *thd, HA_CREATE_INFO *create_info,
             it preparable for open. But let us do close_temporary_table() here
             just in case.
           */
-          drop_temporary_table(thd, create_table);
+          drop_temporary_table(thd, create_table, NULL);
         }
         else
           table= create_table->table;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 57417acdd93..ba508783133 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -2298,7 +2298,7 @@ void Query_tables_list::reset_query_tables_list(bool init)
   sroutines_list_own_last= sroutines_list.next;
   sroutines_list_own_elements= 0;
   binlog_stmt_flags= 0;
-  stmt_accessed_table_flag= 0; 
+  stmt_accessed_table_flag= 0;
 }
 
 
@@ -3143,3 +3143,150 @@ bool LEX::is_partition_management() const
            alter_info.flags == ALTER_REORGANIZE_PARTITION));
 }
 
+
+#ifdef MYSQL_SERVER
+uint binlog_unsafe_map[256];
+
+#define UNSAFE(a, b, c) \
+  { \
+  DBUG_PRINT("unsafe_mixed_statement", ("SETTING BASE VALUES: %s, %s, %02X\n", \
+    LEX::stmt_accessed_table_string(a), \
+    LEX::stmt_accessed_table_string(b), \
+    c)); \
+  unsafe_mixed_statement(a, b, c); \
+  }
+
+/*
+  Sets the combination given by "a" and "b" and automatically combinations
+  given by other types of access, i.e. 2^(8 - 2), as unsafe.
+
+  It may happen a colision when automatically defining a combination as unsafe.
+  For that reason, a combination has its unsafe condition redefined only when
+  the new_condition is greater then the old. For instance,
+  
+     . (BINLOG_DIRECT_ON & TRX_CACHE_NOT_EMPTY) is never overwritten by 
+     . (BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF).
+*/
+void unsafe_mixed_statement(LEX::enum_stmt_accessed_table a,
+                            LEX::enum_stmt_accessed_table b, uint condition)
+{
+  int type= 0;
+  int index= (1U << a) | (1U << b);
+  
+  
+  for (type= 0; type < 256; type++)
+  {
+    if ((type & index) == index)
+    {
+      binlog_unsafe_map[type] |= condition;
+    }
+  }
+}
+/*
+  The BINLOG_* AND TRX_CACHE_* values can be combined by using '&' or '|',
+  which means that both conditions need to be satisfied or any of them is
+  enough. For example, 
+    
+    . BINLOG_DIRECT_ON & TRX_CACHE_NOT_EMPTY means that the statment is
+    unsafe when the option is on and trx-cache is not empty;
+
+    . BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF means the statement is unsafe
+    in all cases.
+
+    . TRX_CACHE_EMPTY | TRX_CACHE_NOT_EMPTY means the statement is unsafe
+    in all cases. Similar as above.
+*/
+void binlog_unsafe_map_init()
+{
+  memset((void*) binlog_unsafe_map, 0, sizeof(uint) * 256);
+
+  /*
+    Classify a statement as unsafe when there is a mixed statement and an
+    on-going transaction at any point of the execution if:
+
+      1. The mixed statement is about to update a transactional table and
+      a non-transactional table.
+
+      2. The mixed statement is about to update a transactional table and
+      read from a non-transactional table.
+
+      3. The mixed statement is about to update a non-transactional table
+      and temporary transactional table.
+
+      4. The mixed statement is about to update a temporary transactional
+      table and read from a non-transactional table.
+
+      5. The mixed statement is about to update a transactional table and
+      a temporary non-transactional table.
+     
+      6. The mixed statement is about to update a transactional table and
+      read from a temporary non-transactional table.
+
+      7. The mixed statement is about to update a temporary transactional
+      table and temporary non-transactional table.
+
+      8. The mixed statement is about to update a temporary transactional
+      table and read from a temporary non-transactional table.
+
+    After updating a transactional table if:
+
+      9. The mixed statement is about to update a non-transactional table
+      and read from a transactional table.
+
+      10. The mixed statement is about to update a non-transactional table
+      and read from a temporary transactional table.
+
+      11. The mixed statement is about to update a temporary non-transactional
+      table and read from a transactional table.
+      
+      12. The mixed statement is about to update a temporary non-transactional
+      table and read from a temporary transactional table.
+
+      13. The mixed statement is about to update a temporary non-transactional
+      table and read from a non-transactional table.
+
+    The reason for this is that locks acquired may not protected a concurrent
+    transaction of interfering in the current execution and by consequence in
+    the result.
+  */
+  /* Case 1. */
+  UNSAFE(LEX::STMT_WRITES_TRANS_TABLE, LEX::STMT_WRITES_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF);
+  /* Case 2. */
+  UNSAFE(LEX::STMT_WRITES_TRANS_TABLE, LEX::STMT_READS_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF);
+  /* Case 3. */
+  UNSAFE(LEX::STMT_WRITES_NON_TRANS_TABLE, LEX::STMT_WRITES_TEMP_TRANS_TABLE,
+    BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF);
+  /* Case 4. */
+  UNSAFE(LEX::STMT_WRITES_TEMP_TRANS_TABLE, LEX::STMT_READS_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF);
+  /* Case 5. */
+  UNSAFE(LEX::STMT_WRITES_TRANS_TABLE, LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON);
+  /* Case 6. */
+  UNSAFE(LEX::STMT_WRITES_TRANS_TABLE, LEX::STMT_READS_TEMP_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON);
+  /* Case 7. */
+  UNSAFE(LEX::STMT_WRITES_TEMP_TRANS_TABLE, LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON);
+  /* Case 8. */
+  UNSAFE(LEX::STMT_WRITES_TEMP_TRANS_TABLE, LEX::STMT_READS_TEMP_NON_TRANS_TABLE,
+    BINLOG_DIRECT_ON);
+  /* Case 9. */
+  UNSAFE(LEX::STMT_WRITES_NON_TRANS_TABLE, LEX::STMT_READS_TRANS_TABLE,
+    (BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF) & TRX_CACHE_NOT_EMPTY);
+  /* Case 10 */
+  UNSAFE(LEX::STMT_WRITES_NON_TRANS_TABLE, LEX::STMT_READS_TEMP_TRANS_TABLE,
+    (BINLOG_DIRECT_ON | BINLOG_DIRECT_OFF) & TRX_CACHE_NOT_EMPTY);
+  /* Case 11. */
+  UNSAFE(LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE, LEX::STMT_READS_TRANS_TABLE,
+    BINLOG_DIRECT_ON & TRX_CACHE_NOT_EMPTY);
+  /* Case 12. */
+  UNSAFE(LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE, LEX::STMT_READS_TEMP_TRANS_TABLE,
+    BINLOG_DIRECT_ON & TRX_CACHE_NOT_EMPTY);
+  /* Case 13. */
+  UNSAFE(LEX::STMT_WRITES_TEMP_NON_TRANS_TABLE, LEX::STMT_READS_NON_TRANS_TABLE,
+     BINLOG_DIRECT_OFF & TRX_CACHE_NOT_EMPTY);
+}
+#endif
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 0eb8e5de6d2..121b7622597 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -47,6 +47,53 @@ class Key;
 class File_parser;
 class Key_part_spec;
 
+#ifdef MYSQL_SERVER
+/*
+  There are 8 different type of table access so there is no more than
+  combinations 2^8 = 256:
+
+  . STMT_READS_TRANS_TABLE
+
+  . STMT_READS_NON_TRANS_TABLE
+
+  . STMT_READS_TEMP_TRANS_TABLE
+
+  . STMT_READS_TEMP_NON_TRANS_TABLE
+
+  . STMT_WRITES_TRANS_TABLE
+
+  . STMT_WRITES_NON_TRANS_TABLE
+
+  . STMT_WRITES_TEMP_TRANS_TABLE
+
+  . STMT_WRITES_TEMP_NON_TRANS_TABLE
+
+  The unsafe conditions for each combination is represented within a byte
+  and stores the status of the option --binlog-direct-non-trans-updates,
+  whether the trx-cache is empty or not, and whether the isolation level
+  is lower than ISO_REPEATABLE_READ:
+
+  . option (OFF/ON)
+  . trx-cache (empty/not empty)
+  . isolation (>= ISO_REPEATABLE_READ / < ISO_REPEATABLE_READ)
+
+  bits 0 : . OFF, . empty, . >= ISO_REPEATABLE_READ
+  bits 1 : . OFF, . empty, . < ISO_REPEATABLE_READ
+  bits 2 : . OFF, . not empty, . >= ISO_REPEATABLE_READ
+  bits 3 : . OFF, . not empty, . < ISO_REPEATABLE_READ
+  bits 4 : . ON, . empty, . >= ISO_REPEATABLE_READ
+  bits 5 : . ON, . empty, . < ISO_REPEATABLE_READ
+  bits 6 : . ON, . not empty, . >= ISO_REPEATABLE_READ
+  bits 7 : . ON, . not empty, . < ISO_REPEATABLE_READ
+*/
+extern uint binlog_unsafe_map[256];
+/*
+  Initializes the array with unsafe combinations and its respective
+  conditions.
+*/
+void binlog_unsafe_map_init();
+#endif
+
 /**
   used by the parser to store internal variable name
 */
@@ -1302,33 +1349,33 @@ public:
     */
     STMT_READS_TRANS_TABLE= 0,
     /*
-       If a transactional table is about to be updated.
-    */
-    STMT_WRITES_TRANS_TABLE,
-    /*
        If a non-transactional table is about to be read. Note that
        a write implies a read.
     */
     STMT_READS_NON_TRANS_TABLE,
     /*
-       If a non-transactional table is about to be updated.
-    */
-    STMT_WRITES_NON_TRANS_TABLE,
-    /*
        If a temporary transactional table is about to be read. Note
        that a write implies a read.
     */
     STMT_READS_TEMP_TRANS_TABLE,
     /*
-       If a temporary transactional table is about to be updated.
-    */
-    STMT_WRITES_TEMP_TRANS_TABLE,
-    /*
        If a temporary non-transactional table is about to be read. Note
       that a write implies a read.
     */
     STMT_READS_TEMP_NON_TRANS_TABLE,
     /*
+       If a transactional table is about to be updated.
+    */
+    STMT_WRITES_TRANS_TABLE,
+    /*
+       If a non-transactional table is about to be updated.
+    */
+    STMT_WRITES_NON_TRANS_TABLE,
+    /*
+       If a temporary transactional table is about to be updated.
+    */
+    STMT_WRITES_TEMP_TRANS_TABLE,
+    /*
        If a temporary non-transactional table is about to be updated.
     */
     STMT_WRITES_TEMP_NON_TRANS_TABLE,
@@ -1338,7 +1385,58 @@ public:
     */
     STMT_ACCESS_TABLE_COUNT
   };
+
+  static inline const char *stmt_accessed_table_string(enum_stmt_accessed_table accessed_table)
+  {
+    switch (accessed_table)
+    {
+      case STMT_READS_TRANS_TABLE:
+         return "STMT_READS_TRANS_TABLE";
+      break;
+      case STMT_READS_NON_TRANS_TABLE:
+        return "STMT_READS_NON_TRANS_TABLE";
+      break;
+      case STMT_READS_TEMP_TRANS_TABLE:
+        return "STMT_READS_TEMP_TRANS_TABLE";
+      break;
+      case STMT_READS_TEMP_NON_TRANS_TABLE:
+        return "STMT_READS_TEMP_NON_TRANS_TABLE";
+      break;  
+      case STMT_WRITES_TRANS_TABLE:
+        return "STMT_WRITES_TRANS_TABLE";
+      break;
+      case STMT_WRITES_NON_TRANS_TABLE:
+        return "STMT_WRITES_NON_TRANS_TABLE";
+      break;
+      case STMT_WRITES_TEMP_TRANS_TABLE:
+        return "STMT_WRITES_TEMP_TRANS_TABLE";
+      break;
+      case STMT_WRITES_TEMP_NON_TRANS_TABLE:
+        return "STMT_WRITES_TEMP_NON_TRANS_TABLE";
+      break;
+      case STMT_ACCESS_TABLE_COUNT:
+      default:
+        DBUG_ASSERT(0);
+      break;
+    }
+  }
                
+  #define BINLOG_DIRECT_ON 0xF0    /* unsafe when
+                                      --binlog-direct-non-trans-updates
+                                      is ON */
+
+  #define BINLOG_DIRECT_OFF 0xF    /* unsafe when
+                                      --binlog-direct-non-trans-updates
+                                      is OFF */
+
+  #define TRX_CACHE_EMPTY 0x33     /* unsafe when trx-cache is empty */
+
+  #define TRX_CACHE_NOT_EMPTY 0xCC /* unsafe when trx-cache is not empty */
+
+  #define IL_LT_REPEATABLE 0xAA    /* unsafe when < ISO_REPEATABLE_READ */
+
+  #define IL_GTE_REPEATABLE 0x55   /* unsafe when >= ISO_REPEATABLE_READ */
+  
   /**
     Sets the type of table that is about to be accessed while executing a
     statement.
@@ -1348,7 +1446,7 @@ public:
   */
   inline void set_stmt_accessed_table(enum_stmt_accessed_table accessed_table)
   {
-    DBUG_ENTER("THD::set_stmt_accessed_table");
+    DBUG_ENTER("LEX::set_stmt_accessed_table");
 
     DBUG_ASSERT(accessed_table >= 0 && accessed_table < STMT_ACCESS_TABLE_COUNT);
     stmt_accessed_table_flag |= (1U << accessed_table);
@@ -1361,7 +1459,7 @@ public:
     statement.
 
     @param accessed_table Enumeration type that defines the type of table,
-                           e.g. temporary, transactional, non-transactional.
+           e.g. temporary, transactional, non-transactional.
 
     @return
       @retval TRUE  if the type of the table is about to be accessed
@@ -1369,7 +1467,7 @@ public:
   */
   inline bool stmt_accessed_table(enum_stmt_accessed_table accessed_table)
   {
-    DBUG_ENTER("THD::stmt_accessed_table");
+    DBUG_ENTER("LEX::stmt_accessed_table");
 
     DBUG_ASSERT(accessed_table >= 0 && accessed_table < STMT_ACCESS_TABLE_COUNT);
 
@@ -1377,40 +1475,79 @@ public:
   }
 
   /**
-    Checks if a temporary table is about to be accessed while executing a
-    statement.
+    Checks if a temporary non-transactional table is about to be accessed
+    while executing a statement.
 
     @return
-      @retval TRUE  if a temporary table is about to be accessed
+      @retval TRUE  if a temporary non-transactional table is about to be
+                    accessed
       @retval FALSE otherwise
   */
-  inline bool stmt_accessed_temp_table()
+  inline bool stmt_accessed_non_trans_temp_table()
   {
-    DBUG_ENTER("THD::stmt_accessed_temp_table");
+    DBUG_ENTER("THD::stmt_accessed_non_trans_temp_table");
 
     DBUG_RETURN((stmt_accessed_table_flag &
-                ((1U << STMT_READS_TEMP_TRANS_TABLE) |
-                 (1U << STMT_WRITES_TEMP_TRANS_TABLE) |
-                 (1U << STMT_READS_TEMP_NON_TRANS_TABLE) |
+                ((1U << STMT_READS_TEMP_NON_TRANS_TABLE) |
                  (1U << STMT_WRITES_TEMP_NON_TRANS_TABLE))) != 0);
   }
 
-  /**
-    Checks if a temporary non-transactional table is about to be accessed
-    while executing a statement.
+  /*
+    Checks if a mixed statement is unsafe.
 
+    
+    @param in_multi_stmt_transaction_mode defines if there is an on-going
+           multi-transactional statement.
+    @param binlog_direct defines if --binlog-direct-non-trans-updates is
+           active.
+    @param trx_cache_is_not_empty defines if the trx-cache is empty or not.
+    @param trx_isolation defines the isolation level.
+ 
     @return
-      @retval TRUE  if a temporary non-transactional table is about to be
-                    accessed
+      @retval TRUE if the mixed statement is unsafe
       @retval FALSE otherwise
   */
-  inline bool stmt_accessed_non_trans_temp_table()
+  inline bool is_mixed_stmt_unsafe(bool in_multi_stmt_transaction_mode,
+                                   bool binlog_direct,
+                                   bool trx_cache_is_not_empty,
+                                   uint tx_isolation)
   {
-    DBUG_ENTER("THD::stmt_accessed_non_trans_temp_table");
+    bool unsafe= FALSE;
 
-    DBUG_RETURN((stmt_accessed_table_flag &
-                ((1U << STMT_READS_TEMP_NON_TRANS_TABLE) |
-                 (1U << STMT_WRITES_TEMP_NON_TRANS_TABLE))) != 0);
+    if (in_multi_stmt_transaction_mode)
+    {
+       uint condition=
+         (binlog_direct ? BINLOG_DIRECT_ON : BINLOG_DIRECT_OFF) &
+         (trx_cache_is_not_empty ? TRX_CACHE_NOT_EMPTY : TRX_CACHE_EMPTY) &
+         (tx_isolation >= ISO_REPEATABLE_READ ? IL_GTE_REPEATABLE : IL_LT_REPEATABLE);
+
+      unsafe= (binlog_unsafe_map[stmt_accessed_table_flag] & condition);
+
+#if !defined(DBUG_OFF)
+      DBUG_PRINT("LEX::is_mixed_stmt_unsafe", ("RESULT %02X %02X %02X\n", condition,
+              binlog_unsafe_map[stmt_accessed_table_flag],
+              (binlog_unsafe_map[stmt_accessed_table_flag] & condition)));
+ 
+      int type_in= 0;
+      for (; type_in < STMT_ACCESS_TABLE_COUNT; type_in++)
+      {
+        if (stmt_accessed_table((enum_stmt_accessed_table) type_in))
+          DBUG_PRINT("LEX::is_mixed_stmt_unsafe", ("ACCESSED %s ",
+                  stmt_accessed_table_string((enum_stmt_accessed_table) type_in)));
+      }
+#endif
+    }
+
+    if (stmt_accessed_table(STMT_WRITES_NON_TRANS_TABLE) &&
+      stmt_accessed_table(STMT_READS_TRANS_TABLE) &&
+      tx_isolation < ISO_REPEATABLE_READ)
+      unsafe= TRUE;
+    else if (stmt_accessed_table(STMT_WRITES_TEMP_NON_TRANS_TABLE) &&
+      stmt_accessed_table(STMT_READS_TRANS_TABLE) &&
+      tx_isolation < ISO_REPEATABLE_READ)
+      unsafe= TRUE;
+
+    return(unsafe);
   }
 
   /**
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 224de005af3..de492a14c78 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -1795,6 +1795,8 @@ end:
     clear_error                   is clear_error to be called
     query                         Query to log
     query_length                  Length of query
+    is_trans                      if the event changes either
+                                  a trans or non-trans engine.
 
   RETURN VALUES
     NONE
@@ -1917,18 +1919,64 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
   String wrong_tables;
   int error= 0;
   int non_temp_tables_count= 0;
-  bool some_tables_deleted=0, tmp_table_deleted=0, foreign_key_error=0;
+  bool foreign_key_error=0;
+  bool non_tmp_error= 0;
+  bool trans_tmp_table_deleted= 0, non_trans_tmp_table_deleted= 0;
+  bool non_tmp_table_deleted= 0;
   String built_query;
-  String built_tmp_query;
+  String built_trans_tmp_query, built_non_trans_tmp_query;
   DBUG_ENTER("mysql_rm_table_part2");
 
-  if (thd->is_current_stmt_binlog_format_row() && !dont_log_query)
+  /*
+    Prepares the drop statements that will be written into the binary
+    log as follows:
+
+    1 - If we are not processing a "DROP TEMPORARY" it prepares a
+    "DROP".
+
+    2 - A "DROP" may result in a "DROP TEMPORARY" but the opposite is
+    not true.
+
+    3 - If the current format is row, the IF EXISTS token needs to be
+    appended because one does not know if CREATE TEMPORARY was previously
+    written to the binary log.
+
+    4 - Add the IF_EXISTS token if necessary, i.e. if_exists is TRUE.
+
+    5 - For temporary tables, there is a need to differentiate tables
+    in transactional and non-transactional storage engines. For that,
+    reason, two types of drop statements are prepared.
+
+    The need to different the type of tables when dropping a temporary
+    table stems from the fact that such drop does not commit an ongoing
+    transaction and changes to non-transactional tables must be written
+    ahead of the transaction in some circumstances.
+  */
+  if (!dont_log_query)
   {
-    built_query.set_charset(system_charset_info);
-    if (if_exists)
-      built_query.append("DROP TABLE IF EXISTS ");
+    if (!drop_temporary)
+    {
+      built_query.set_charset(system_charset_info);
+      if (if_exists)
+        built_query.append("DROP TABLE IF EXISTS ");
+      else
+        built_query.append("DROP TABLE ");
+    }
+
+    if (thd->is_current_stmt_binlog_format_row() || if_exists)
+    {
+      built_trans_tmp_query.set_charset(system_charset_info);
+      built_trans_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
+      built_non_trans_tmp_query.set_charset(system_charset_info);
+      built_non_trans_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
+    }
     else
-      built_query.append("DROP TABLE ");
+    {
+      built_trans_tmp_query.set_charset(system_charset_info);
+      built_trans_tmp_query.append("DROP TEMPORARY TABLE ");
+      built_non_trans_tmp_query.set_charset(system_charset_info);
+      built_non_trans_tmp_query.append("DROP TEMPORARY TABLE ");
+    }
   }
 
   mysql_ha_rm_tables(thd, tables);
@@ -1997,6 +2045,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 
   for (table= tables; table; table= table->next_local)
   {
+    bool is_trans;
     char *db=table->db;
     handlerton *table_type;
     enum legacy_db_type frm_db_type= DB_TYPE_UNKNOWN;
@@ -2005,78 +2054,70 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                          table->db, table->table_name, (long) table->table,
                          table->table ? (long) table->table->s : (long) -1));
 
+    /*
+      drop_temporary_table may return one of the following error codes:
+      .  0 - a temporary table was successfully dropped.
+      .  1 - a temporary table was not found.
+      . -1 - a temporary table is used by an outer statement.
+    */
     if (table->open_type == OT_BASE_ONLY)
       error= 1;
-    else
-      error= drop_temporary_table(thd, table);
- 
-    switch (error) {
-    case  0:
-      // removed temporary table
-      tmp_table_deleted= 1;
+    else if ((error= drop_temporary_table(thd, table, &is_trans)) == -1)
+    {
+      DBUG_ASSERT(thd->in_sub_stmt);
+      goto err;
+    }
+
+    if ((drop_temporary && if_exists) || !error)
+    {
       /*
-        One needs to always log any temporary table drop if the current
-        statement logging format is set to row. This happens because one
-        might have created a temporary table while the statement logging
-        format was statement and then switched to mixed or row format.
+        This handles the case of temporary tables. We have the following cases:
+
+          . "DROP TEMPORARY" was executed and a temporary table was affected
+          (i.e. drop_temporary && !error) or the if_exists was specified (i.e.
+          drop_temporary && if_exists).
+
+          . "DROP" was executed but a temporary table was affected (.i.e
+          !error).
       */
-      if (thd->is_current_stmt_binlog_format_row())
+      if (!dont_log_query)
       {
-        if (built_tmp_query.is_empty()) 
-        {
-          built_tmp_query.set_charset(system_charset_info);
-          built_tmp_query.append("DROP TEMPORARY TABLE IF EXISTS ");
-        }
+        /*
+          If there is an error, we don't know the type of the engine
+          at this point. So, we keep it in the trx-cache.
+        */
+        is_trans= error ? TRUE : is_trans;
+        if (is_trans)
+          trans_tmp_table_deleted= TRUE;
+        else
+          non_trans_tmp_table_deleted= TRUE;
 
-        built_tmp_query.append("`");
+        String *built_ptr_query=
+          (is_trans ? &built_trans_tmp_query : &built_non_trans_tmp_query);
+        /*
+          Don't write the database name if it is the current one (or if
+          thd->db is NULL).
+        */
+        built_ptr_query->append("`");
         if (thd->db == NULL || strcmp(db,thd->db) != 0)
         {
-          built_tmp_query.append(db);
-          built_tmp_query.append("`.`");
+          built_ptr_query->append(db);
+          built_ptr_query->append("`.`");
         }
-        built_tmp_query.append(table->table_name);
-        built_tmp_query.append("`,");
+        built_ptr_query->append(table->table_name);
+        built_ptr_query->append("`,");
       }
-
-      continue;
-    case -1:
-      DBUG_ASSERT(thd->in_sub_stmt);
-      error= 1;
-      goto err;
-    default:
-      // temporary table not found
-      error= 0;
-    }
-
-    /* Probably a non-temporary table. */
-    if (!drop_temporary)
-      non_temp_tables_count++;
-
-    /*
-      If row-based replication is used and the table is not a
-      temporary table, we add the table name to the drop statement
-      being built.  The string always end in a comma and the comma
-      will be chopped off before being written to the binary log.
-      */
-    if (!drop_temporary && thd->is_current_stmt_binlog_format_row() && !dont_log_query)
-    {
       /*
-        Don't write the database name if it is the current one (or if
-        thd->db is NULL).
+        This means that a temporary table was droped and as such there
+        is no need to proceed with the code that tries to drop a regular
+        table.
       */
-      built_query.append("`");
-      if (thd->db == NULL || strcmp(db,thd->db) != 0)
-      {
-        built_query.append(db);
-        built_query.append("`.`");
-      }
-
-      built_query.append(table->table_name);
-      built_query.append("`,");
+      if (!error) continue;
     }
-
-    if (!drop_temporary)
+    else if (!drop_temporary)
     {
+      non_temp_tables_count++;
+
       if (thd->locked_tables_mode)
       {
         if (wait_while_table_is_used(thd, table->table, HA_EXTRA_FORCE_REOPEN))
@@ -2099,7 +2140,38 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                                         reg_ext,
                                         table->internal_tmp_table ?
                                         FN_IS_TMP : 0);
+
+      /*
+        This handles the case where a "DROP" was executed and a regular
+        table "may be" dropped as drop_temporary is FALSE and error is
+        TRUE. If the error was FALSE a temporary table was dropped and
+        regardless of the status of drop_tempoary a "DROP TEMPORARY"
+        must be used.
+      */
+      if (!dont_log_query)
+      {
+        /*
+          Note that unless if_exists is TRUE or a temporary table was deleted, 
+          there is no means to know if the statement should be written to the
+          binary log. See further information on this variable in what follows.
+        */
+        non_tmp_table_deleted= (if_exists ? TRUE : non_tmp_table_deleted);
+        /*
+          Don't write the database name if it is the current one (or if
+          thd->db is NULL).
+        */
+        built_query.append("`");
+        if (thd->db == NULL || strcmp(db,thd->db) != 0)
+        {
+          built_query.append(db);
+          built_query.append("`.`");
+        }
+
+        built_query.append(table->table_name);
+        built_query.append("`,");
+      }
     }
+
     /*
       TODO: Investigate what should be done to remove this lock completely.
             Is exclusive meta-data lock enough ?
@@ -2108,19 +2180,29 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     DBUG_EXECUTE_IF("sleep_before_part2_delete_table",
                     my_sleep(100000););
     mysql_mutex_lock(&LOCK_open);
+    error= 0;
     if (drop_temporary ||
         ((access(path, F_OK) &&
           ha_create_table_from_engine(thd, db, alias)) ||
          (!drop_view &&
           dd_frm_type(thd, path, &frm_db_type) != FRMTYPE_TABLE)))
     {
-      // Table was not found on disk and table can't be created from engine
+      /*
+        One of the following cases happened:
+          . "DROP TEMPORARY" but a temporary table was not found.
+          . "DROP" but table was not found on disk and table can't be
+            created from engine.
+          . ./sql/datadict.cc +32 /Alfranio - TODO: We need to test this.
+      */
       if (if_exists)
 	push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
 			    ER_BAD_TABLE_ERROR, ER(ER_BAD_TABLE_ERROR),
 			    table->table_name);
       else
+      {
+        non_tmp_error = (drop_temporary ? non_tmp_error : TRUE);
         error= 1;
+      }
     }
     else
     {
@@ -2153,7 +2235,7 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
       if (error == HA_ERR_ROW_IS_REFERENCED)
       {
 	/* the table is referenced by a foreign key constraint */
-	foreign_key_error=1;
+	foreign_key_error= 1;
       }
       if (!error || error == ENOENT || error == HA_ERR_NO_SUCH_TABLE)
       {
@@ -2162,12 +2244,13 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
 	strmov(end,reg_ext);
         if (!(new_error= mysql_file_delete(key_file_frm, path, MYF(MY_WME))))
         {
-	  some_tables_deleted=1;
+          non_tmp_table_deleted= TRUE;
           new_error= Table_triggers_list::drop_all_triggers(thd, db,
                                                             table->table_name);
         }
         error|= new_error;
       }
+       non_tmp_error= error ? TRUE : non_tmp_error;
     }
     mysql_mutex_unlock(&LOCK_open);
     if (error)
@@ -2183,11 +2266,12 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
                     my_printf_error(ER_BAD_TABLE_ERROR,
                                     ER(ER_BAD_TABLE_ERROR), MYF(0),
                                     table->table_name););
-
   }
   DEBUG_SYNC(thd, "rm_table_part2_before_binlog");
-  thd->thread_specific_used|= tmp_table_deleted;
+  thd->thread_specific_used|= (trans_tmp_table_deleted ||
+                               non_trans_tmp_table_deleted);
   error= 0;
+err:
   if (wrong_tables.length())
   {
     if (!foreign_key_error)
@@ -2198,85 +2282,48 @@ int mysql_rm_table_part2(THD *thd, TABLE_LIST *tables, bool if_exists,
     error= 1;
   }
 
-  if (some_tables_deleted || tmp_table_deleted || !error)
+  if (non_trans_tmp_table_deleted ||
+      trans_tmp_table_deleted || non_tmp_table_deleted)
   {
     query_cache_invalidate3(thd, tables, 0);
     if (!dont_log_query && mysql_bin_log.is_open())
     {
-      if (!thd->is_current_stmt_binlog_format_row() ||
-          (non_temp_tables_count > 0 && !tmp_table_deleted))
+      if (non_trans_tmp_table_deleted)
       {
-        /*
-          In this case, we are either using statement-based
-          replication or using row-based replication but have only
-          deleted one or more non-temporary tables (and no temporary
-          tables).  In this case, we can write the original query into
-          the binary log.
-        */
-        error |= write_bin_log(thd, !error, thd->query(), thd->query_length());
+          /* Chop of the last comma */
+          built_non_trans_tmp_query.chop();
+          built_non_trans_tmp_query.append(" /* generated by server */");
+          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                     built_non_trans_tmp_query.ptr(),
+                                     built_non_trans_tmp_query.length(),
+                                     FALSE, FALSE, FALSE, 0);
       }
-      else if (thd->is_current_stmt_binlog_format_row() &&
-               tmp_table_deleted)
+      if (trans_tmp_table_deleted)
       {
-        if (non_temp_tables_count > 0)
-        {
-          /*
-            In this case we have deleted both temporary and
-            non-temporary tables, so:
-            - since we have deleted a non-temporary table we have to
-              binlog the statement, but
-            - since we have deleted a temporary table we cannot binlog
-              the statement (since the table may have not been created on the
-              slave - check "if" branch below, this might cause the slave to 
-              stop).
-
-            Instead, we write a built statement, only containing the
-            non-temporary tables, to the binary log
-          */
-          built_query.chop();                  // Chop of the last comma
+          /* Chop of the last comma */
+          built_trans_tmp_query.chop();
+          built_trans_tmp_query.append(" /* generated by server */");
+          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                     built_trans_tmp_query.ptr(),
+                                     built_trans_tmp_query.length(),
+                                     TRUE, FALSE, FALSE, 0);
+      }
+      if (non_tmp_table_deleted)
+      {
+          /* Chop of the last comma */
+          built_query.chop();
           built_query.append(" /* generated by server */");
-          error|= write_bin_log(thd, !error, built_query.ptr(), built_query.length());
-        }
-
-        /*
-          One needs to always log any temporary table drop if the current
-          statement logging format is set to row. This happens because one
-          might have created a temporary table while the statement logging
-          format was statement and then switched to mixed or row format.
-        */
-        if (thd->is_current_stmt_binlog_format_row())
-        {
-          /*
-            In this case we have deleted some temporary tables but we are using
-            row based logging for the statement. However, thread uses mixed mode
-            format, thence we need to log the dropping as we cannot tell for
-            sure whether the create was logged as statement previously or not, ie,
-            before switching to row mode.
-          */
-          built_tmp_query.chop();                  // Chop of the last comma
-          built_tmp_query.append(" /* generated by server */");
-          /*
-            We cannot call the write_bin_log as we do not care about any errors
-            in the master as the statement is always DROP TEMPORARY TABLE IF EXISTS
-            and as such there will be no errors in the slave.
-          */
-          error|= thd->binlog_query(THD::STMT_QUERY_TYPE, built_tmp_query.ptr(),
-                                    built_tmp_query.length(), FALSE, FALSE, FALSE,
-                                    0);
-        }
+          int error_code = (non_tmp_error ?
+            (foreign_key_error ? ER_ROW_IS_REFERENCED : ER_BAD_TABLE_ERROR) : 0);
+          error |= thd->binlog_query(THD::STMT_QUERY_TYPE,
+                                     built_query.ptr(),
+                                     built_query.length(),
+                                     TRUE, FALSE, FALSE,
+                                     error_code);
       }
-
-      /*
-        The remaining cases are:
-        - no tables were deleted and
-        - only temporary tables were deleted and row-based
-          replication is used.
-        In both these cases, nothing should be written to the binary
-        log.
-      */
     }
   }
-err:
+
   if (!drop_temporary)
   {
     /*
@@ -3791,6 +3838,8 @@ void sp_prepare_create_field(THD *thd, Create_field *sql_field)
     internal_tmp_table  Set to 1 if this is an internal temporary table
 			(From ALTER TABLE)
     select_field_count
+    is_trans            identifies the type of engine where the table
+                        was created: either trans or non-trans.
 
   DESCRIPTION
     If one creates a temporary table, this is automatically opened
@@ -3815,7 +3864,8 @@ bool mysql_create_table_no_lock(THD *thd,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info,
                                 bool internal_tmp_table,
-                                uint select_field_count)
+                                uint select_field_count,
+                                bool *is_trans)
 {
   char		path[FN_REFLEN + 1];
   uint          path_length;
@@ -4180,12 +4230,17 @@ bool mysql_create_table_no_lock(THD *thd,
 
   if (create_info->options & HA_LEX_CREATE_TMP_TABLE)
   {
+    TABLE *table= NULL;
     /* Open table and put in temporary table list */
-    if (!(open_temporary_table(thd, path, db, table_name, 1)))
+    if (!(table= open_temporary_table(thd, path, db, table_name, 1)))
     {
       (void) rm_temporary_table(create_info->db_type, path);
       goto unlock_and_end;
     }
+
+    if (is_trans != NULL)
+      *is_trans= table->file->has_transactions();
+
     thd->thread_specific_used= TRUE;
   }
 
@@ -4236,9 +4291,10 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
   /* Got lock. */
   DEBUG_SYNC(thd, "locked_table_name");
 
+  bool is_trans;
   result= mysql_create_table_no_lock(thd, create_table->db,
                                      create_table->table_name, create_info,
-                                     alter_info, FALSE, 0);
+                                     alter_info, FALSE, 0, &is_trans);
 
   /*
     Don't write statement if:
@@ -4250,7 +4306,7 @@ bool mysql_create_table(THD *thd, TABLE_LIST *create_table,
       (!thd->is_current_stmt_binlog_format_row() ||
        (thd->is_current_stmt_binlog_format_row() &&
         !(create_info->options & HA_LEX_CREATE_TMP_TABLE))))
-    result= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
+    result= write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans);
 
 end:
   DBUG_RETURN(result);
@@ -4456,9 +4512,10 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
   /* Reset auto-increment counter for the new table. */
   local_create_info.auto_increment_value= 0;
 
+  bool is_trans;
   if ((res= mysql_create_table_no_lock(thd, table->db, table->table_name,
                                        &local_create_info, &local_alter_info,
-                                       FALSE, 0)))
+                                       FALSE, 0, &is_trans)))
     goto err;
 
   /*
@@ -4539,7 +4596,7 @@ bool mysql_create_like_table(THD* thd, TABLE_LIST* table, TABLE_LIST* src_table,
       Case 3 and 4 does nothing under RBR
     */
   }
-  else if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
+  else if (write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans))
     goto err;
 
 err:
@@ -6228,7 +6285,7 @@ bool mysql_alter_table(THD *thd,char *new_db, char *new_name,
   error= mysql_create_table_no_lock(thd, new_db, tmp_name,
                                     create_info,
                                     alter_info,
-                                    1, 0);
+                                    1, 0, NULL);
   reenable_binlog(thd);
   if (error)
     goto err;
diff --git a/sql/sql_table.h b/sql/sql_table.h
index 2e1cc89d585..ae5beefea37 100644
--- a/sql/sql_table.h
+++ b/sql/sql_table.h
@@ -138,7 +138,8 @@ bool mysql_create_table_no_lock(THD *thd, const char *db,
                                 const char *table_name,
                                 HA_CREATE_INFO *create_info,
                                 Alter_info *alter_info,
-                                bool tmp_table, uint select_field_count);
+                                bool tmp_table, uint select_field_count,
+                                bool *is_trans);
 bool mysql_prepare_alter_table(THD *thd, TABLE *table,
                                HA_CREATE_INFO *create_info,
                                Alter_info *alter_info);