BUG#50479 DDL stmt on row-only/stmt-only tables generate spurious binlog_format

errors
            
In the fix of BUG#39934 in 5.1-rep+3, errors are generated when
binlog_format=row and a statement modifies a table restricted to
statement-logging (ER_BINLOG_ROW_MODE_AND_STMT_ENGINE); or if
binlog_format=statement and a statement modifies a table restricted to
row-logging (ER_BINLOG_STMT_MODE_AND_ROW_ENGINE).
            
However, some DDL statements that lock tables (e.g. ALTER TABLE,
CREATE INDEX and CREATE TRIGGER) were causing spurious errors,
although no row might be inserted into the binary log.
            
To fix the problem, we tagged statements that may generate
rows into the binary log and thence the warning messages are
only printed out when the appropriate conditions hold and rows
might be changed.

sql/log_event.cc:
  Reorganized the Query_log_event's constructor based on the
  CF_CAN_GENERATE_ROW_EVENTS flag and as such any statement
  that has the associated flag should go through a cache
  before being written to the binary log.
sql/share/errmsg-utf8.txt:
  Improved the error message ER_BINLOG_UNSAFE_MIXED_STATEMENT according to Paul's
  suggestion.
sql/sql_class.cc:
  Created a hook to be used by innodb that checks if a statement
  may write rows to the binary log. In other words, if it has
  the CF_CAN_GENERATE_ROW_EVENTS flag associated.
sql/sql_class.h:
  Defined the CF_CAN_GENERATE_ROW_EVENTS flag.
sql/sql_parse.cc:
  Updated the sql_command_flags and added a function to check the 
  CF_CAN_GENERATE_ROW_EVENTS.
sql/sql_parse.h:
  Added a function to check the CF_CAN_GENERATE_ROW_EVENTS.
storage/innobase/handler/ha_innodb.cc:
  Added a call to the hook thd_generates_rows().
storage/innobase/handler/ha_innodb.h:
  Defined an external reference to the hook thd_generates_rows().

6 files changed, 96 insertions, 87 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 00015ea52fe..f7c6d09f98f 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2465,95 +2465,62 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg,
   else
     time_zone_len= 0;
 
+  LEX *lex= thd->lex;
   /*
-    In what follows, we decide whether to write to the binary log or to use a
-    cache.
+    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.
+
+    Note that a cache will not be used if the parameter direct is
+    TRUE.
   */
-  LEX *lex= thd->lex;
-  bool implicit_commit= FALSE;
-  bool force_trans= FALSE;
+  bool use_cache= FALSE;
+  /*
+    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.
+  */
+  bool trx_cache= FALSE;
   cache_type= Log_event::EVENT_INVALID_CACHE;
+
   switch (lex->sql_command)
   {
-    case SQLCOM_ALTER_DB:
-    case SQLCOM_CREATE_FUNCTION:
-    case SQLCOM_DROP_FUNCTION:
-    case SQLCOM_DROP_PROCEDURE:
-    case SQLCOM_INSTALL_PLUGIN:
-    case SQLCOM_UNINSTALL_PLUGIN:
-    case SQLCOM_ALTER_TABLESPACE:
-      implicit_commit= TRUE;
-      break;
     case SQLCOM_DROP_TABLE:
-      force_trans= lex->drop_temporary && thd->in_multi_stmt_transaction_mode();
-      implicit_commit= !force_trans;
-      break;
-    case SQLCOM_ALTER_TABLE:
+      use_cache= trx_cache= (lex->drop_temporary &&
+                            thd->in_multi_stmt_transaction_mode());
+    break;
+
     case SQLCOM_CREATE_TABLE:
-      force_trans= (lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) &&
-                    thd->in_multi_stmt_transaction_mode();
-      implicit_commit= !force_trans &&
-                       !(lex->select_lex.item_list.elements &&
-                         thd->is_current_stmt_binlog_format_row());
+      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 &&
+                  thd->is_current_stmt_binlog_format_row());
       break;
     case SQLCOM_SET_OPTION:
-      implicit_commit= (lex->autocommit ? TRUE : FALSE);
-      break;
-    /*
-      Replace what follows after CF_AUTO_COMMIT_TRANS is backported by:
-
-      default:
-        implicit_commit= ((sql_command_flags[lex->sql_command] &
-                      CF_AUTO_COMMIT_TRANS));
+      use_cache= trx_cache= (lex->autocommit ? FALSE : TRUE);
       break;
-    */
-    case SQLCOM_CREATE_INDEX:
-    case SQLCOM_TRUNCATE:
-    case SQLCOM_CREATE_DB:
-    case SQLCOM_DROP_DB:
-    case SQLCOM_ALTER_DB_UPGRADE:
-    case SQLCOM_RENAME_TABLE:
-    case SQLCOM_DROP_INDEX:
-    case SQLCOM_CREATE_VIEW:
-    case SQLCOM_DROP_VIEW:
-    case SQLCOM_CREATE_TRIGGER:
-    case SQLCOM_DROP_TRIGGER:
-    case SQLCOM_CREATE_EVENT:
-    case SQLCOM_ALTER_EVENT:
-    case SQLCOM_DROP_EVENT:
-    case SQLCOM_REPAIR:
-    case SQLCOM_OPTIMIZE:
-    case SQLCOM_ANALYZE:
-    case SQLCOM_CREATE_USER:
-    case SQLCOM_DROP_USER:
-    case SQLCOM_RENAME_USER:
-    case SQLCOM_REVOKE_ALL:
-    case SQLCOM_REVOKE:
-    case SQLCOM_GRANT:
-    case SQLCOM_CREATE_PROCEDURE:
-    case SQLCOM_CREATE_SPFUNCTION:
-    case SQLCOM_ALTER_PROCEDURE:
-    case SQLCOM_ALTER_FUNCTION:
-    case SQLCOM_ASSIGN_TO_KEYCACHE:
-    case SQLCOM_PRELOAD_KEYS:
-    case SQLCOM_FLUSH:
-    case SQLCOM_RESET:
-    case SQLCOM_CHECK:
-      implicit_commit= TRUE;
+    case SQLCOM_RELEASE_SAVEPOINT:
+    case SQLCOM_ROLLBACK_TO_SAVEPOINT:
+    case SQLCOM_SAVEPOINT:
+      use_cache= trx_cache= TRUE;
       break;
     default:
-      implicit_commit= FALSE;
+      use_cache= sqlcom_can_generate_row_events(thd);
       break;
   }
 
-  if (implicit_commit || direct)
+  if (!use_cache || direct)
   {
     cache_type= Log_event::EVENT_NO_CACHE;
   }
   else
   {
-    cache_type= ((using_trans || stmt_has_updated_trans_table(thd) ||
-                 force_trans || thd->thread_temporary_used)
+    cache_type= ((using_trans || stmt_has_updated_trans_table(thd)
+                  || trx_cache || thd->thread_temporary_used)
                  ? Log_event::EVENT_TRANSACTIONAL_CACHE :
                  Log_event::EVENT_STMT_CACHE);
   }
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index f6ed6330749..9c2cb40badc 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -6335,7 +6335,7 @@ ER_BINLOG_UNSAFE_MULTIPLE_ENGINES_AND_SELF_LOGGING_ENGINE
   eng "Mixing self-logging and non-self-logging engines in a statement is unsafe."
 
 ER_BINLOG_UNSAFE_MIXED_STATEMENT
-  eng "Statements that read from both transactional (or a temporary table of any engine type) and non-transactional tables and write to any of them are unsafe."
+  eng "Statement accesses nontransactional table as well as transactional or temporary table, and writes to any of them."
 
 ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SQL_LOG_BIN
   eng "Cannot modify @@session.sql_log_bin inside a transaction"
@@ -6343,4 +6343,3 @@ ER_INSIDE_TRANSACTION_PREVENTS_SWITCH_SQL_LOG_BIN
 ER_STORED_FUNCTION_PREVENTS_SWITCH_SQL_LOG_BIN
   eng "Cannot change the sql_log_bin inside a stored function or trigger"
 
-
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index dfe17fe5779..b090f35a607 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -3192,6 +3192,11 @@ extern "C" bool thd_binlog_filter_ok(const MYSQL_THD thd)
 {
   return binlog_filter->db_ok(thd->db);
 }
+
+extern "C" bool thd_sqlcom_can_generate_row_events(const MYSQL_THD thd)
+{
+  return sqlcom_can_generate_row_events(thd);
+}
 #endif // INNODB_COMPATIBILITY_HOOKS */
 
 /****************************************************************************
@@ -3917,7 +3922,8 @@ int THD::decide_logging_format(TABLE_LIST *tables)
         */
         my_error((error= ER_BINLOG_ROW_INJECTION_AND_STMT_ENGINE), MYF(0));
       }
-      else if (variables.binlog_format == BINLOG_FORMAT_ROW)
+      else if (variables.binlog_format == BINLOG_FORMAT_ROW &&
+               sqlcom_can_generate_row_events(this))
       {
         /*
           2. Error: Cannot modify table that uses a storage engine
@@ -3955,7 +3961,8 @@ int THD::decide_logging_format(TABLE_LIST *tables)
           */
           my_error((error= ER_BINLOG_ROW_INJECTION_AND_STMT_MODE), MYF(0));
         }
-        else if ((flags_write_all_set & HA_BINLOG_STMT_CAPABLE) == 0)
+        else if ((flags_write_all_set & HA_BINLOG_STMT_CAPABLE) == 0 &&
+                 sqlcom_can_generate_row_events(this))
         {
           /*
             5. Error: Cannot modify table that uses a storage engine
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 15aef33bcb3..015a87cb5cc 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -3619,6 +3619,12 @@ public:
  */
 #define CF_PROTECT_AGAINST_GRL  (1U << 10)
 
+/**
+  Identifies statements that may generate row events
+  and that may end up in the binary log.
+*/
+#define CF_CAN_GENERATE_ROW_EVENTS (1U << 11)
+
 /* Bits in server_command_flags */
 
 /**
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 90d0a52d40d..06bebe76842 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -247,8 +247,19 @@ void init_update_queries(void)
   /* Initialize the sql command flags array. */
   memset(sql_command_flags, 0, sizeof(sql_command_flags));
 
+  /*
+    In general, DDL statements do not generate row events and do not go
+    through a cache before being written to the binary log. However, the
+    CREATE TABLE...SELECT is an exception because it may generate row
+    events. For that reason,  the SQLCOM_CREATE_TABLE  which represents
+    a CREATE TABLE, including the CREATE TABLE...SELECT, has the
+    CF_CAN_GENERATE_ROW_EVENTS flag. The distinction between a regular
+    CREATE TABLE and the CREATE TABLE...SELECT is made in other parts of
+    the code, in particular in the Query_log_event's constructor.
+  */
   sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
+                                            CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND |
                                             CF_AUTO_COMMIT_TRANS | CF_PROTECT_AGAINST_GRL;
@@ -256,7 +267,8 @@ void init_update_queries(void)
                                             CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA | CF_AUTO_COMMIT_TRANS;
   sql_command_flags[SQLCOM_ALTER_DB_UPGRADE]= CF_AUTO_COMMIT_TRANS;
@@ -275,22 +287,32 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_DROP_TRIGGER]=   CF_AUTO_COMMIT_TRANS;
 
   sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_UPDATE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_DELETE_MULTI]=   CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
-                                            CF_PROTECT_AGAINST_GRL;
-  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
-  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE;
-  sql_command_flags[SQLCOM_SELECT]=         CF_REEXECUTION_FRAGILE;
+                                            CF_PROTECT_AGAINST_GRL |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_SELECT]=         CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
   sql_command_flags[SQLCOM_SET_OPTION]=     CF_REEXECUTION_FRAGILE | CF_AUTO_COMMIT_TRANS;
-  sql_command_flags[SQLCOM_DO]=             CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_DO]=             CF_REEXECUTION_FRAGILE |
+                                            CF_CAN_GENERATE_ROW_EVENTS;
 
   sql_command_flags[SQLCOM_SHOW_STATUS_PROC]= CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
   sql_command_flags[SQLCOM_SHOW_STATUS]=      CF_STATUS_COMMAND | CF_REEXECUTION_FRAGILE;
@@ -365,7 +387,9 @@ void init_update_queries(void)
     last called (or executed) statement is preserved.
     See mysql_execute_command() for how CF_ROW_COUNT is used.
   */
-  sql_command_flags[SQLCOM_CALL]=      CF_REEXECUTION_FRAGILE;
+  sql_command_flags[SQLCOM_CALL]=      CF_REEXECUTION_FRAGILE |
+                                       CF_CAN_GENERATE_ROW_EVENTS;
+  sql_command_flags[SQLCOM_EXECUTE]=   CF_CAN_GENERATE_ROW_EVENTS;
 
   /*
     The following admin table operations are allowed
@@ -390,7 +414,12 @@ void init_update_queries(void)
   sql_command_flags[SQLCOM_CHECK]=              CF_AUTO_COMMIT_TRANS;
 }
 
-
+bool sqlcom_can_generate_row_events(const THD *thd)
+{
+  return (sql_command_flags[thd->lex->sql_command] &
+          CF_CAN_GENERATE_ROW_EVENTS);
+}
+ 
 bool is_update_query(enum enum_sql_command command)
 {
   DBUG_ASSERT(command >= 0 && command <= SQLCOM_END);
diff --git a/sql/sql_parse.h b/sql/sql_parse.h
index e1543a09549..6d968033ccd 100644
--- a/sql/sql_parse.h
+++ b/sql/sql_parse.h
@@ -79,6 +79,7 @@ bool check_host_name(LEX_STRING *str);
 bool check_identifier_name(LEX_STRING *str, uint max_char_length,
                            uint err_code, const char *param_for_err_msg);
 bool mysql_test_parse_for_slave(THD *thd,char *inBuf,uint length);
+bool sqlcom_can_generate_row_events(const THD *thd);
 bool is_update_query(enum enum_sql_command command);
 bool is_log_table_write_query(enum enum_sql_command command);
 bool alloc_query(THD *thd, const char *packet, uint packet_length);