BUG#23051 (READ COMMITTED breaks mixed and statement-based

replication):

Patch to add binlog format capabilities to the InnoDB storage engine.
The engine will not allow statement format logging when in READ COMMITTED
or READ UNCOMMITTED transaction isolation level.

In addition, an error is generated when trying to use READ COMMITTED
or READ UNCOMMITTED transaction isolation level in STATEMENT binlog
mode.

mysql-test/r/binlog_innodb.result

+SET BINLOG_FORMAT=MIXED;
+RESET MASTER;
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=INNODB;
+INSERT INTO t1 VALUES (1,1),(2,2),(3,3),(4,4),(5,5),(6,6);
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 2*a WHERE a > 1;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+UPDATE t1 SET b = a * a WHERE a > 3;
+COMMIT;
+SET BINLOG_FORMAT=STATEMENT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+UPDATE t1 SET b = 1*a WHERE a > 1;
+ERROR HY000: Logging not possible. Message: InnoDB: Transaction level 'READ-UNCOMMITTED' is not safe for binlog mode 'STATEMENT'
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+UPDATE t1 SET b = 2*a WHERE a > 2;
+ERROR HY000: Logging not possible. Message: InnoDB: Transaction level 'READ-COMMITTED' is not safe for binlog mode 'STATEMENT'
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 3*a WHERE a > 3;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+UPDATE t1 SET b = 4*a WHERE a > 4;
+COMMIT;
+SET BINLOG_FORMAT=MIXED;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+UPDATE t1 SET b = 1*a WHERE a > 1;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+UPDATE t1 SET b = 2*a WHERE a > 2;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 3*a WHERE a > 3;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+UPDATE t1 SET b = 4*a WHERE a > 4;
+COMMIT;
+SET BINLOG_FORMAT=ROW;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+UPDATE t1 SET b = 1*a WHERE a > 1;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+UPDATE t1 SET b = 2*a WHERE a > 2;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 3*a WHERE a > 3;
+COMMIT;
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+UPDATE t1 SET b = 4*a WHERE a > 4;
+COMMIT;
+show binlog events from <binlog_start>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=INNODB
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1 VALUES (1,1),(2,2),(3,3),(4,4),(5,5),(6,6)
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1 SET b = 2*a WHERE a > 1
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1 SET b = 3*a WHERE a > 3
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1 SET b = 4*a WHERE a > 4
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1 SET b = 3*a WHERE a > 3
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1 SET b = 4*a WHERE a > 4
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+master-bin.000001	#	Query	#	#	use `test`; BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Xid	#	#	COMMIT /* XID */
+DROP TABLE t1;

mysql-test/t/binlog_innodb.test

+source include/have_innodb.inc;
+
+SET BINLOG_FORMAT=MIXED;
+
+RESET MASTER;
+
+CREATE TABLE t1 (a INT PRIMARY KEY, b INT) ENGINE=INNODB;
+INSERT INTO t1 VALUES (1,1),(2,2),(3,3),(4,4),(5,5),(6,6);
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+# Should be logged as statement
+UPDATE t1 SET b = 2*a WHERE a > 1;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+# Should be logged as rows
+UPDATE t1 SET b = a * a WHERE a > 3;
+COMMIT;
+
+# Check that errors are generated when trying to use READ COMMITTED
+# transaction isolation level in STATEMENT binlog mode.
+
+SET BINLOG_FORMAT=STATEMENT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+error ER_BINLOG_LOGGING_IMPOSSIBLE;
+UPDATE t1 SET b = 1*a WHERE a > 1;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+error ER_BINLOG_LOGGING_IMPOSSIBLE;
+UPDATE t1 SET b = 2*a WHERE a > 2;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 3*a WHERE a > 3;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+UPDATE t1 SET b = 4*a WHERE a > 4;
+COMMIT;
+
+SET BINLOG_FORMAT=MIXED;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+UPDATE t1 SET b = 1*a WHERE a > 1;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+UPDATE t1 SET b = 2*a WHERE a > 2;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 3*a WHERE a > 3;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+UPDATE t1 SET b = 4*a WHERE a > 4;
+COMMIT;
+
+SET BINLOG_FORMAT=ROW;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
+UPDATE t1 SET b = 1*a WHERE a > 1;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED;
+UPDATE t1 SET b = 2*a WHERE a > 2;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+UPDATE t1 SET b = 3*a WHERE a > 3;
+COMMIT;
+
+BEGIN;
+SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE;
+UPDATE t1 SET b = 4*a WHERE a > 4;
+COMMIT;
+
+source include/show_binlog_events.inc;
+
+DROP TABLE t1;

sql/handler.h

@@ -1708,6 +1708,8 @@ private:
 	/* Some extern variables used with handlers */
 
 extern const char *ha_row_type[];
+extern const char *tx_isolation_names[];
+extern const char *binlog_format_names[];
 extern TYPELIB tx_isolation_typelib;
 extern TYPELIB myisam_stats_method_typelib;
 extern ulong total_ha, total_ha_2pc;

sql/share/errmsg.txt

@@ -6062,9 +6062,5 @@ ER_NO_PARTITION_FOR_GIVEN_VALUE_SILENT
 ER_BINLOG_UNSAFE_STATEMENT
         eng "Statement is not safe to log in statement format."
         swe "Detta r inte skert att logga i statement-format."
-ER_BINLOG_ENGINES_INCOMPATIBLE
-        eng "It is not possible to log anything with this combination of engines"
-ER_BINLOG_STMT_FORMAT_FORBIDDEN
-        eng "Attempting to log statement in in statement format, but statement format is not possible with this combination of engines"
-ER_BINLOG_ROW_FORMAT_FORBIDDEN
-        eng "Attempting to log statement in in row format, but row format is not possible with this combination of engines"
+ER_BINLOG_LOGGING_IMPOSSIBLE
+        eng "Binary logging not possible. Message: %s"

sql/sql_base.cc

@@ -3612,18 +3612,24 @@ int decide_logging_format(THD *thd, TABLE_LIST *tables)
     int error= 0;
     if (binlog_flags == 0)
     {
-      error= ER_BINLOG_ENGINES_INCOMPATIBLE;
+      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 &&
              (binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
     {
-      error= ER_BINLOG_STMT_FORMAT_FORBIDDEN;
+      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()) &&
              (binlog_flags & HA_BINLOG_ROW_CAPABLE) == 0)
     {
-      error= ER_BINLOG_ROW_FORMAT_FORBIDDEN;
+      my_error((error= ER_BINLOG_LOGGING_IMPOSSIBLE), MYF(0),
+                "Row-based format required for this statement,"
+                " but not allowed by this combination of engines");
     }
 
     DBUG_PRINT("info", ("error: %d", error));
@@ -3631,7 +3637,6 @@ int decide_logging_format(THD *thd, TABLE_LIST *tables)
     if (error)
     {
       ha_rollback_stmt(thd);
-      my_error(error, MYF(0));
       return -1;
     }
 

storage/innobase/handler/ha_innodb.cc

@@ -1003,6 +1003,7 @@ ha_innobase::ha_innobase(handlerton *hton, TABLE_SHARE *table_arg)
 		  HA_CAN_SQL_HANDLER |
 		  HA_PRIMARY_KEY_REQUIRED_FOR_POSITION |
 		  HA_PRIMARY_KEY_IN_READ_INDEX |
+                  HA_BINLOG_ROW_CAPABLE |
 		  HA_CAN_GEOMETRY | HA_PARTIAL_COLUMN_READ |
 		  HA_TABLE_SCAN_ON_INDEX),
   start_of_scan(0),
@@ -2314,6 +2315,45 @@ ha_innobase::get_row_type() const
 	return(ROW_TYPE_NOT_USED);
 }
 
+
+
+/********************************************************************
+Get the table flags to use for the statement. */
+handler::Table_flags
+ha_innobase::table_flags() const
+{
+	THD *const thd= current_thd;
+        /* We are using thd->variables.tx_isolation here instead of
+           trx->isolation_level since store_lock() has not been called
+           yet.
+
+           The trx->isolation_level is set inside store_lock() (which
+           is called from mysql_lock_tables()) until after this
+           function has been called (which is called in lock_tables()
+           before that function calls mysql_lock_tables()). */
+        ulong const tx_isolation= thd->variables.tx_isolation;
+        if (tx_isolation <= ISO_READ_COMMITTED)
+        {
+	        ulong const binlog_format= thd->variables.binlog_format;
+                /* Statement based binlogging does not work in these
+                   isolation levels since the necessary locks cannot
+                   be taken */
+        	if (binlog_format == BINLOG_FORMAT_STMT)
+          	{
+			char buf[256];
+	                my_snprintf(buf, sizeof(buf),
+                                    "Transaction level '%s' in InnoDB is"
+                                    " not safe for binlog mode '%s'",
+                                    tx_isolation_names[tx_isolation],
+                                    binlog_format_names[binlog_format]);
+                        my_error(ER_BINLOG_LOGGING_IMPOSSIBLE, MYF(0), buf);
+                }
+                return int_table_flags;
+        }
+
+        return int_table_flags | HA_BINLOG_STMT_CAPABLE;
+}
+
 /********************************************************************
 Gives the file extension of an InnoDB single-table tablespace. */
 static const char* ha_innobase_exts[] = {

storage/innobase/handler/ha_innodb.h

@@ -54,7 +54,7 @@ class ha_innobase: public handler
 	ulong		upd_and_key_val_buff_len;
 					/* the length of each of the previous
 					two buffers */
-	ulong		int_table_flags;
+	Table_flags	int_table_flags;
 	uint		primary_key;
 	ulong		start_of_scan;	/* this is set to 1 when we are
 					starting a table scan but have not
@@ -84,7 +84,7 @@ class ha_innobase: public handler
 	const char* table_type() const { return("InnoDB");}
 	const char *index_type(uint key_number) { return "BTREE"; }
 	const char** bas_ext() const;
-	ulonglong table_flags() const { return int_table_flags; }
+	Table_flags table_flags() const;
 	ulong index_flags(uint idx, uint part, bool all_parts) const
 	{
 	  return (HA_READ_NEXT |