WL#3303 (RBR: Engine-controlled logging format):

Adding support to allow engines to tell what formats they can handle.
The server will generate an error if it is not possible to log the
statement according to the logging mode in effect.

Adding flags to several storage engines to state what they can handle.

Changes to NDB handler removing code that forces row-based mode and
adding flag saying that NDB can only handle row format.

Adding check that binlog flags are only used for real tables that are
opened for writing.


BitKeeper/deleted/.del-binlog_row_blackhole.result:
  Rename: mysql-test/r/binlog_row_blackhole.result -> BitKeeper/deleted/.del-binlog_row_blackhole.result
BitKeeper/deleted/.del-binlog_row_blackhole.test:
  Rename: mysql-test/t/binlog_row_blackhole.test -> BitKeeper/deleted/.del-binlog_row_blackhole.test
mysql-test/t/partition_hash.test:
  Adding error check for statement that might fail.
sql/ha_ndbcluster.cc:
  Removing statements that switch to row-based format.
  Adding row capabilities.
sql/handler.h:
  Adding handler/table flags to indicate that the engine is row- and/or
  statement-logging capable.
  
  Adding typedef for table_flags type.
sql/set_var.cc:
  Removing code that prevents changing binlog format when NDB is active.
sql/share/errmsg.txt:
  Adding error messages for when row- and/or statement-based logging
  formats cannot be used.
sql/sql_base.cc:
  Adding business logic in lock_tables() to decide when an error should
  be thrown because logging is not possible.
  Add logic to switch to row format when that is allowed and needed.
  ---
  Binlog flags should only be checked for real tables that are opened for
  writing. Adding code to check that.
storage/archive/ha_archive.h:
  Adding row- and statement-logging capabilities to engine.
storage/blackhole/ha_blackhole.h:
  Blackhole can handle statement-format only.
storage/csv/ha_tina.h:
  Adding row- and statement-logging capabilities to engine.
storage/example/ha_example.h:
  For the example engine, we arbitrarily decided that it only can handle
  row format.
storage/federated/ha_federated.h:
  Adding row- and statement-logging capabilities to engine.
storage/heap/ha_heap.h:
  Heap can handle both row- and statement-based logging format.
storage/myisam/ha_myisam.cc:
  MyISAM can handle both row- and statement-based logging format.
storage/myisammrg/ha_myisammrg.h:
  MyISAM can handle both row- and statement-based logging format.
mysql-test/r/binlog_multi_engine.result:
  New BitKeeper file ``mysql-test/r/binlog_multi_engine.result''
mysql-test/t/binlog_multi_engine.test:
  New BitKeeper file ``mysql-test/t/binlog_multi_engine.test''

18 files changed, 271 insertions, 188 deletions

diff --git a/mysql-test/r/binlog_multi_engine.result b/mysql-test/r/binlog_multi_engine.result
new file mode 100644
index 00000000000..7af4525d887
--- /dev/null
+++ b/mysql-test/r/binlog_multi_engine.result
@@ -0,0 +1,86 @@
+CREATE TABLE t1m (m INT, n INT) ENGINE=MYISAM;
+CREATE TABLE t1b (b INT, c INT) ENGINE=BLACKHOLE;
+CREATE TABLE t1n (e INT, f INT) ENGINE=NDB;
+SET SESSION BINLOG_FORMAT=STATEMENT;
+INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2);
+ERROR HY000: Attempting to log statement in in statement format, but statement format is not possible with this combination of engines
+UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c;
+UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f;
+ERROR HY000: Attempting to log statement in in statement format, but statement format is not possible with this combination of engines
+UPDATE t1n, t1b SET e = 2, b = 3 WHERE f = c;
+ERROR HY000: It is not possible to log anything with this combination of engines
+TRUNCATE t1m;
+TRUNCATE t1b;
+TRUNCATE t1n;
+SET SESSION BINLOG_FORMAT=MIXED;
+INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2);
+UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c;
+UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f;
+UPDATE t1n, t1b SET e = 2, b = 3 WHERE f = c;
+ERROR HY000: It is not possible to log anything with this combination of engines
+TRUNCATE t1m;
+TRUNCATE t1b;
+TRUNCATE t1n;
+SET SESSION BINLOG_FORMAT=ROW;
+INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2);
+ERROR HY000: Attempting to log statement in in row format, but row format is not possible with this combination of engines
+INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2);
+UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c;
+ERROR HY000: Attempting to log statement in in row format, but row format is not possible with this combination of engines
+UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f;
+UPDATE t1n, t1b SET e = 2, b = 3 WHERE f = c;
+ERROR HY000: It is not possible to log anything with this combination of engines
+TRUNCATE t1m;
+TRUNCATE t1b;
+TRUNCATE t1n;
+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 t1m (m INT, n INT) ENGINE=MYISAM
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1b (b INT, c INT) ENGINE=BLACKHOLE
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1n (e INT, f INT) ENGINE=NDB
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2)
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1m
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1b
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1n
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2)
+master-bin.000001	#	Query	#	#	use `test`; INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2)
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c
+master-bin.000001	#	Query	#	#	use `test`; UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1m
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1b
+master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1n)
+master-bin.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
+master-bin.000001	#	Write_rows	#	#	table_id: #
+master-bin.000001	#	Write_rows	#	#	table_id: #
+master-bin.000001	#	Write_rows	#	#	table_id: #
+master-bin.000001	#	Update_rows	#	#	table_id: #
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1n
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1m)
+master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1m)
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1n)
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1m
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1b
+master-bin.000001	#	Query	#	#	BEGIN
+master-bin.000001	#	Table_map	#	#	table_id: # (test.t1n)
+master-bin.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
+master-bin.000001	#	Write_rows	#	#	table_id: #
+master-bin.000001	#	Write_rows	#	#	table_id: #
+master-bin.000001	#	Write_rows	#	#	table_id: #
+master-bin.000001	#	Update_rows	#	#	table_id: #
+master-bin.000001	#	Update_rows	#	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	#	#	COMMIT
+master-bin.000001	#	Query	#	#	use `test`; TRUNCATE t1n
+DROP TABLE t1m, t1b, t1n;
diff --git a/mysql-test/r/binlog_row_blackhole.result b/mysql-test/r/binlog_row_blackhole.result
deleted file mode 100644
index 8e90ac4f30b..00000000000
--- a/mysql-test/r/binlog_row_blackhole.result
+++ /dev/null
@@ -1,152 +0,0 @@
-drop table if exists t1,t2;
-CREATE TABLE t1 (
-Period smallint(4) unsigned zerofill DEFAULT '0000' NOT NULL,
-Varor_period smallint(4) unsigned DEFAULT '0' NOT NULL
-) ENGINE=blackhole;
-INSERT INTO t1 VALUES (9410,9412);
-select period from t1;
-period
-select * from t1;
-Period	Varor_period
-select t1.* from t1;
-Period	Varor_period
-CREATE TABLE t2 (
-auto int NOT NULL auto_increment,
-fld1 int(6) unsigned zerofill DEFAULT '000000' NOT NULL,
-companynr tinyint(2) unsigned zerofill DEFAULT '00' NOT NULL,
-fld3 char(30) DEFAULT '' NOT NULL,
-fld4 char(35) DEFAULT '' NOT NULL,
-fld5 char(35) DEFAULT '' NOT NULL,
-fld6 char(4) DEFAULT '' NOT NULL,
-primary key (auto)
-) ENGINE=blackhole;
-INSERT INTO t2 VALUES (1192,068305,00,'Colombo','hardware','colicky','');
-INSERT INTO t2 VALUES (1193,000000,00,'nondecreasing','implant','thrillingly','');
-select t2.fld3 from t2 where companynr = 58 and fld3 like "%imaginable%";
-fld3
-select fld3 from t2 where fld3 like "%cultivation" ;
-fld3
-select t2.fld3,companynr from t2 where companynr = 57+1 order by fld3;
-fld3	companynr
-select fld3,companynr from t2 where companynr = 58 order by fld3;
-fld3	companynr
-select fld3 from t2 order by fld3 desc limit 10;
-fld3
-select fld3 from t2 order by fld3 desc limit 5;
-fld3
-select fld3 from t2 order by fld3 desc limit 5,5;
-fld3
-select t2.fld3 from t2 where fld3 = 'honeysuckle';
-fld3
-select t2.fld3 from t2 where fld3 LIKE 'honeysuckl_';
-fld3
-select t2.fld3 from t2 where fld3 LIKE 'hon_ysuckl_';
-fld3
-select t2.fld3 from t2 where fld3 LIKE 'honeysuckle%';
-fld3
-select t2.fld3 from t2 where fld3 LIKE 'h%le';
-fld3
-select t2.fld3 from t2 where fld3 LIKE 'honeysuckle_';
-fld3
-select t2.fld3 from t2 where fld3 LIKE 'don_t_find_me_please%';
-fld3
-select t2.fld3 from t2 where fld3 >= 'honeysuckle' and fld3 <= 'honoring' order by fld3;
-fld3
-select fld1,fld3 from t2 where fld3="Colombo" or fld3 = "nondecreasing" order by fld3;
-fld1	fld3
-DROP TABLE t1;
-CREATE TABLE t1 (a VARCHAR(200), b TEXT, FULLTEXT (a,b));
-INSERT INTO t1 VALUES('MySQL has now support', 'for full-text search'),
-('Full-text indexes', 'are called collections'),
-('Only MyISAM tables','support collections'),
-('Function MATCH ... AGAINST()','is used to do a search'),
-('Full-text search in MySQL', 'implements vector space model');
-SHOW INDEX FROM t1;
-Table	Non_unique	Key_name	Seq_in_index	Column_name	Collation	Cardinality	Sub_part	Packed	Null	Index_type	Comment
-t1	1	a	1	a	NULL	NULL	NULL	NULL	YES	FULLTEXT	
-t1	1	a	2	b	NULL	NULL	NULL	NULL	YES	FULLTEXT	
-select * from t1 where MATCH(a,b) AGAINST ("collections");
-a	b
-Only MyISAM tables	support collections
-Full-text indexes	are called collections
-explain extended select * from t1 where MATCH(a,b) AGAINST ("collections");
-id	select_type	table	type	possible_keys	key	key_len	ref	rows	filtered	Extra
-1	SIMPLE	t1	fulltext	a	a	0		1	100.00	Using where
-Warnings:
-Note	1003	select `test`.`t1`.`a` AS `a`,`test`.`t1`.`b` AS `b` from `test`.`t1` where (match `test`.`t1`.`a`,`test`.`t1`.`b` against (_latin1'collections'))
-select * from t1 where MATCH(a,b) AGAINST ("indexes");
-a	b
-Full-text indexes	are called collections
-select * from t1 where MATCH(a,b) AGAINST ("indexes collections");
-a	b
-Full-text indexes	are called collections
-Only MyISAM tables	support collections
-select * from t1 where MATCH(a,b) AGAINST ("only");
-a	b
-reset master;
-drop table t1,t2;
-create table t1 (a int) engine=blackhole;
-delete from t1 where a=10;
-update t1 set a=11 where a=15;
-insert into t1 values(1);
-insert ignore into t1 values(1);
-replace into t1 values(100);
-create table t2 (a varchar(200)) engine=blackhole;
-load data infile '../std_data_ln/words.dat' into table t2;
-alter table t1 add b int;
-alter table t1 drop b;
-create table t3 like t1;
-insert into t1 select * from t3;
-replace into t1 select * from t3;
-select * from t1;
-a
-select * from t2;
-a
-select * from t3;
-a
-show binlog events from <binlog_start>;
-Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	use `test`; drop table t1,t2
-master-bin.000001	#	Query	#	#	use `test`; create table t1 (a int) engine=blackhole
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin.000001	#	Query	#	#	use `test`; create table t2 (a varchar(200)) engine=blackhole
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin.000001	#	Query	#	#	use `test`; alter table t1 add b int
-master-bin.000001	#	Query	#	#	use `test`; alter table t1 drop b
-master-bin.000001	#	Query	#	#	use `test`; create table t3 like t1
-drop table t1,t2,t3;
-CREATE TABLE t1(a INT, b INT) ENGINE=BLACKHOLE;
-DELETE FROM t1 WHERE a=10;
-ALTER TABLE t1 ADD INDEX(a);
-DELETE FROM t1 WHERE a=10;
-ALTER TABLE t1 DROP INDEX a;
-ALTER TABLE t1 ADD UNIQUE INDEX(a);
-DELETE FROM t1 WHERE a=10;
-ALTER TABLE t1 DROP INDEX a;
-ALTER TABLE t1 ADD PRIMARY KEY(a);
-DELETE FROM t1 WHERE a=10;
-DROP TABLE t1;
-reset master;
-create table t1 (a int) engine=blackhole;
-set autocommit=0;
-start transaction;
-insert into t1 values(1);
-commit;
-start transaction;
-insert into t1 values(2);
-rollback;
-set autocommit=1;
-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) engine=blackhole
-master-bin.000001	#	Query	#	#	use `test`; BEGIN
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin.000001	#	Query	#	#	use `test`; COMMIT
-drop table if exists t1;
diff --git a/mysql-test/t/binlog_multi_engine.test b/mysql-test/t/binlog_multi_engine.test
new file mode 100644
index 00000000000..25c99ff6ca8
--- /dev/null
+++ b/mysql-test/t/binlog_multi_engine.test
@@ -0,0 +1,59 @@
+source include/have_blackhole.inc;
+source include/have_ndb.inc;
+
+CREATE TABLE t1m (m INT, n INT) ENGINE=MYISAM;
+CREATE TABLE t1b (b INT, c INT) ENGINE=BLACKHOLE;
+CREATE TABLE t1n (e INT, f INT) ENGINE=NDB;
+
+SET SESSION BINLOG_FORMAT=STATEMENT;
+
+INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2);
+error ER_BINLOG_STMT_FORMAT_FORBIDDEN;
+INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2);
+
+UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c;
+error ER_BINLOG_STMT_FORMAT_FORBIDDEN;
+UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f;
+error ER_BINLOG_ENGINES_INCOMPATIBLE;
+UPDATE t1n, t1b SET e = 2, b = 3 WHERE f = c;
+
+TRUNCATE t1m;
+TRUNCATE t1b;
+TRUNCATE t1n;
+
+SET SESSION BINLOG_FORMAT=MIXED;
+
+INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2);
+
+UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c;
+UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f;
+error ER_BINLOG_ENGINES_INCOMPATIBLE;
+UPDATE t1n, t1b SET e = 2, b = 3 WHERE f = c;
+
+TRUNCATE t1m;
+TRUNCATE t1b;
+TRUNCATE t1n;
+
+SET SESSION BINLOG_FORMAT=ROW;
+
+INSERT INTO t1m VALUES (1,1), (1,2), (2,1), (2,2);
+error ER_BINLOG_ROW_FORMAT_FORBIDDEN;
+INSERT INTO t1b VALUES (1,1), (1,2), (2,1), (2,2);
+INSERT INTO t1n VALUES (1,1), (1,2), (2,1), (2,2);
+
+error ER_BINLOG_ROW_FORMAT_FORBIDDEN;
+UPDATE t1m, t1b SET m = 2, b = 3 WHERE n = c;
+UPDATE t1m, t1n SET m = 2, e = 3 WHERE n = f;
+error ER_BINLOG_ENGINES_INCOMPATIBLE;
+UPDATE t1n, t1b SET e = 2, b = 3 WHERE f = c;
+
+TRUNCATE t1m;
+TRUNCATE t1b;
+TRUNCATE t1n;
+
+source include/show_binlog_events.inc;
+
+DROP TABLE t1m, t1b, t1n;
diff --git a/mysql-test/t/binlog_row_blackhole.test b/mysql-test/t/binlog_row_blackhole.test
deleted file mode 100644
index d5355ad1ff0..00000000000
--- a/mysql-test/t/binlog_row_blackhole.test
+++ /dev/null
@@ -1,11 +0,0 @@
-# This is a wrapper for binlog.test so that the same test case can be used 
-# For both statement and row based bin logs 9/19/2005 [jbm]
-
--- source include/have_binlog_format_row.inc
-
-# Bug#18326: Do not lock table for writing during prepare of statement
-# The use of the ps protocol causes extra table maps in the binlog, so
-# we disable the ps-protocol for this statement.
---disable_ps_protocol
--- source extra/binlog_tests/blackhole.test
---enable_ps_protocol
diff --git a/mysql-test/t/partition_hash.test b/mysql-test/t/partition_hash.test
index d3f1a5f4892..fd7628778d8 100644
--- a/mysql-test/t/partition_hash.test
+++ b/mysql-test/t/partition_hash.test
@@ -130,6 +130,7 @@ drop table t1;
 # BUG# 14524 Partitions: crash if blackhole
 #
 CREATE TABLE t1 (s1 int) ENGINE=BLACKHOLE PARTITION BY HASH (s1);
+--error 0,ER_BINLOG_ROW_FORMAT_FORBIDDEN
 INSERT INTO t1 VALUES (0);
 DROP TABLE t1;
 
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 58d34d99174..6fcaf43df34 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -4444,7 +4444,6 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       DBUG_PRINT("warning", ("ops_pending != 0L"));
     m_ops_pending= 0;
   }
-  thd->set_current_stmt_binlog_row_based_if_mixed();
   DBUG_RETURN(error);
 }
 
@@ -4494,7 +4493,6 @@ int ha_ndbcluster::start_stmt(THD *thd, thr_lock_type lock_type)
   m_active_trans= trans;
   // Start of statement
   m_ops_pending= 0;    
-  thd->set_current_stmt_binlog_row_based_if_mixed();
 
   DBUG_RETURN(error);
 }
@@ -6000,6 +5998,7 @@ void ha_ndbcluster::get_auto_increment(ulonglong offset, ulonglong increment,
                 HA_PRIMARY_KEY_REQUIRED_FOR_DELETE | \
                 HA_PARTIAL_COLUMN_READ | \
                 HA_HAS_OWN_BINLOGGING | \
+                HA_BINLOG_ROW_CAPABLE | \
                 HA_HAS_RECORDS
 
 ha_ndbcluster::ha_ndbcluster(handlerton *hton, TABLE_SHARE *table_arg):
@@ -8788,7 +8787,6 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
   my_net_init(&thd->net, 0);
   thd->main_security_ctx.master_access= ~0;
   thd->main_security_ctx.priv_user = 0;
-  thd->current_stmt_binlog_row_based= TRUE;     // If in mixed mode
 
   /* Signal successful initialization */
   ndb_util_thread_running= 1;
diff --git a/sql/handler.h b/sql/handler.h
index cfa86358fa1..21261dc41fa 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -117,6 +117,18 @@
 #define HA_HAS_RECORDS	       (LL(1) << 32) /* records() gives exact count*/
 /* Has it's own method of binlog logging */
 #define HA_HAS_OWN_BINLOGGING  (LL(1) << 33)
+/*
+  Engine is capable of row-format and statement-format logging,
+  respectively
+*/
+#define HA_BINLOG_ROW_CAPABLE  (LL(1) << 34)
+#define HA_BINLOG_STMT_CAPABLE (LL(1) << 35)
+
+/*
+  Set of all binlog flags. Currently only contain the capabilities
+  flags.
+ */
+#define HA_BINLOG_FLAGS (HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE)
 
 /* bits in index_flags(index_number) for what you can do with index */
 #define HA_READ_NEXT            1       /* TODO really use this flag */
@@ -687,7 +699,7 @@ struct handlerton
 };
 
 
-/* Possible flags of a handlerton */
+/* Possible flags of a handlerton (there can be 32 of them) */
 #define HTON_NO_FLAGS                 0
 #define HTON_CLOSE_CURSORS_AT_COMMIT (1 << 0)
 #define HTON_ALTER_NOT_SUPPORTED     (1 << 1) //Engine does not support alter
@@ -889,10 +901,13 @@ class handler :public Sql_alloc
 {
   friend class ha_partition;
 
+public:
+  typedef ulonglong Table_flags;
+
  protected:
   struct st_table_share *table_share;   /* The table definition */
   struct st_table *table;               /* The current open table */
-  ulonglong cached_table_flags;         /* Set on init() and open() */
+  Table_flags cached_table_flags;       /* Set on init() and open() */
 
   virtual int index_init(uint idx, bool sorted) { active_index=idx; return 0; }
   virtual int index_end() { active_index=MAX_KEY; return 0; }
@@ -905,7 +920,7 @@ class handler :public Sql_alloc
   */
   virtual int rnd_init(bool scan) =0;
   virtual int rnd_end() { return 0; }
-  virtual ulonglong table_flags(void) const =0;
+  virtual Table_flags table_flags(void) const =0;
   void ha_statistic_increment(ulong SSV::*offset) const;
 
   ha_rows estimation_rows_to_insert;
@@ -1115,7 +1130,7 @@ public:
   {
     return inited == INDEX ? ha_index_end() : inited == RND ? ha_rnd_end() : 0;
   }
-  longlong ha_table_flags() { return cached_table_flags; }
+  Table_flags ha_table_flags() const { return cached_table_flags; }
 
   /*
     Signal that the table->read_set and table->write_set table maps changed
diff --git a/sql/set_var.cc b/sql/set_var.cc
index 325167ff9fa..a24b3ea049d 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -1343,16 +1343,6 @@ bool sys_var_thd_binlog_format::is_readonly() const
     my_error(ER_STORED_FUNCTION_PREVENTS_SWITCH_BINLOG_FORMAT, MYF(0));
     return 1;    
   }
-#ifdef HAVE_NDB_BINLOG
-  /*
-    Cluster does not support changing the binlog format on the fly yet.
-  */
-  if (opt_bin_log && (have_ndbcluster == SHOW_OPTION_YES))
-  {
-    my_error(ER_NDB_CANT_SWITCH_BINLOG_FORMAT, MYF(0));
-    return 1;
-  }
-#endif /* HAVE_NDB_BINLOG */
   return sys_var_thd_enum::is_readonly();
 }
 
diff --git a/sql/share/errmsg.txt b/sql/share/errmsg.txt
index 2e1240ab616..675ef5bc7b0 100644
--- a/sql/share/errmsg.txt
+++ b/sql/share/errmsg.txt
@@ -6062,3 +6062,9 @@ 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 säkert 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"
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 0ec6ababab0..e2fd5ab65ce 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3585,14 +3585,93 @@ int lock_tables(THD *thd, TABLE_LIST *tables, uint count, bool *need_reopen)
     in prelocked mode.
   */
   DBUG_ASSERT(!thd->prelocked_mode || !thd->lex->requires_prelocking());
-
   *need_reopen= FALSE;
 
-  /*
-    CREATE ... SELECT UUID() locks no tables, we have to test here.
-  */
-  if (thd->lex->is_stmt_unsafe())
-    thd->set_current_stmt_binlog_row_based_if_mixed();
+  if (mysql_bin_log.is_open() && (thd->options | OPTION_BIN_LOG))
+  {
+    /*
+      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.
+    */
+    handler::Table_flags binlog_flags= ~handler::Table_flags();
+    DBUG_PRINT("info", ("HA_BINLOG_FLAGS: 0x%0llx",
+                        (ulonglong) HA_BINLOG_FLAGS));
+    for (table= tables; table; table= table->next_global)
+      if (!table->placeholder() && table->lock_type >= TL_WRITE_ALLOW_WRITE)
+      {
+        DBUG_PRINT("info", ("table: %s; ha_table_flags: 0x%0llx",
+                            table->table_name,
+                            (ulonglong) table->table->file->ha_table_flags()));
+        binlog_flags &= table->table->file->ha_table_flags();
+      }
+    binlog_flags&= HA_BINLOG_FLAGS;
+    DBUG_PRINT("info", ("binlog_flags: 0x%0llx", (ulonglong) binlog_flags));
+    DBUG_PRINT("info", ("thd->variables.binlog_format: %ld",
+                        thd->variables.binlog_format));
+
+    /*
+      We have three alternatives that prevent the statement from being
+      loggable:
+
+      1. If there are no capabilities left (all flags are clear) it is
+         not possible to log the statement at all, so we roll back the
+         statement and report an error.
+
+      2. Statement mode is set, but the capabilities indicate that
+         statement format is not possible.
+
+      3. Row mode is set, but the capabilities indicate that row
+         format is not possible.
+
+      4. Statement is unsafe, but the capabilities indicate that row
+         format is not possible.
+    */
+    int error= 0;
+    if (binlog_flags == 0)
+    {
+      error= ER_BINLOG_ENGINES_INCOMPATIBLE;
+    }
+    else if (thd->variables.binlog_format == BINLOG_FORMAT_STMT &&
+             (binlog_flags & HA_BINLOG_STMT_CAPABLE) == 0)
+    {
+      error= ER_BINLOG_STMT_FORMAT_FORBIDDEN;
+    }
+    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;
+    }
+
+    DBUG_PRINT("info", ("error: %d", error));
+
+    if (error)
+    {
+      ha_rollback_stmt(thd);
+      my_error(error, MYF(0));
+      DBUG_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() ||
+        (binlog_flags | HA_BINLOG_STMT_CAPABLE) == 0)
+    {
+      thd->set_current_stmt_binlog_row_based_if_mixed();
+    }
+  }
 
   if (!tables && !thd->lex->requires_prelocking())
     DBUG_RETURN(0);
diff --git a/storage/archive/ha_archive.h b/storage/archive/ha_archive.h
index 8fc54f6715f..b2ee9221bdd 100644
--- a/storage/archive/ha_archive.h
+++ b/storage/archive/ha_archive.h
@@ -87,6 +87,7 @@ public:
   ulonglong table_flags() const
   {
     return (HA_NO_TRANSACTIONS | HA_REC_NOT_IN_SEQ | HA_CAN_BIT_FIELD |
+            HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
             HA_FILE_BASED | HA_CAN_INSERT_DELAYED | HA_CAN_GEOMETRY);
   }
   ulong index_flags(uint idx, uint part, bool all_parts) const
diff --git a/storage/blackhole/ha_blackhole.h b/storage/blackhole/ha_blackhole.h
index 2af12b33077..bb12c9e04fd 100644
--- a/storage/blackhole/ha_blackhole.h
+++ b/storage/blackhole/ha_blackhole.h
@@ -42,6 +42,7 @@ public:
   ulonglong table_flags() const
   {
     return(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
+           HA_BINLOG_STMT_CAPABLE |
            HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
            HA_FILE_BASED | HA_CAN_GEOMETRY | HA_CAN_INSERT_DELAYED);
   }
diff --git a/storage/csv/ha_tina.h b/storage/csv/ha_tina.h
index c096f21fca2..5d2d05b72c0 100644
--- a/storage/csv/ha_tina.h
+++ b/storage/csv/ha_tina.h
@@ -99,7 +99,8 @@ public:
   const char **bas_ext() const;
   ulonglong table_flags() const
   {
-    return (HA_NO_TRANSACTIONS | HA_REC_NOT_IN_SEQ | HA_NO_AUTO_INCREMENT);
+    return (HA_NO_TRANSACTIONS | HA_REC_NOT_IN_SEQ | HA_NO_AUTO_INCREMENT |
+            HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE);
   }
   ulong index_flags(uint idx, uint part, bool all_parts) const
   {
diff --git a/storage/example/ha_example.h b/storage/example/ha_example.h
index 9777a478209..97ecfb97536 100644
--- a/storage/example/ha_example.h
+++ b/storage/example/ha_example.h
@@ -82,7 +82,12 @@ public:
   */
   ulonglong table_flags() const
   {
-    return 0;
+    /*
+      We are saying that this engine is just row capable to have an
+      engine that can only handle row-based logging. This is used in
+      testing.
+    */
+    return HA_BINLOG_ROW_CAPABLE;
   }
 
   /** @brief
diff --git a/storage/federated/ha_federated.h b/storage/federated/ha_federated.h
index 4d2eefdd986..a6626b2e33e 100644
--- a/storage/federated/ha_federated.h
+++ b/storage/federated/ha_federated.h
@@ -128,6 +128,7 @@ public:
     /* fix server to be able to get remote server table flags */
     return (HA_PRIMARY_KEY_IN_READ_INDEX | HA_FILE_BASED
             | HA_REC_NOT_IN_SEQ | HA_AUTO_PART_KEY | HA_CAN_INDEX_BLOBS |
+            HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
             HA_NO_PREFIX_CHAR_KEYS | HA_PRIMARY_KEY_REQUIRED_FOR_DELETE |
             HA_PARTIAL_COLUMN_READ | HA_NULL_IN_KEY);
   }
diff --git a/storage/heap/ha_heap.h b/storage/heap/ha_heap.h
index a2d531fc515..476efc83666 100644
--- a/storage/heap/ha_heap.h
+++ b/storage/heap/ha_heap.h
@@ -48,6 +48,7 @@ public:
   ulonglong table_flags() const
   {
     return (HA_FAST_KEY_READ | HA_NO_BLOBS | HA_NULL_IN_KEY |
+            HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
             HA_REC_NOT_IN_SEQ | HA_CAN_INSERT_DELAYED | HA_NO_TRANSACTIONS |
             HA_HAS_RECORDS | HA_STATS_RECORDS_IS_EXACT);
   }
diff --git a/storage/myisam/ha_myisam.cc b/storage/myisam/ha_myisam.cc
index d31f5b7e792..61303aaf0ae 100644
--- a/storage/myisam/ha_myisam.cc
+++ b/storage/myisam/ha_myisam.cc
@@ -474,6 +474,7 @@ void mi_check_print_warning(MI_CHECK *param, const char *fmt,...)
 ha_myisam::ha_myisam(handlerton *hton, TABLE_SHARE *table_arg)
   :handler(hton, table_arg), file(0),
   int_table_flags(HA_NULL_IN_KEY | HA_CAN_FULLTEXT | HA_CAN_SQL_HANDLER |
+                  HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
                   HA_DUPLICATE_POS | HA_CAN_INDEX_BLOBS | HA_AUTO_PART_KEY |
                   HA_FILE_BASED | HA_CAN_GEOMETRY | HA_NO_TRANSACTIONS |
                   HA_CAN_INSERT_DELAYED | HA_CAN_BIT_FIELD | HA_CAN_RTREEKEYS |
diff --git a/storage/myisammrg/ha_myisammrg.h b/storage/myisammrg/ha_myisammrg.h
index 7bbe659d4b7..bb22ed13982 100644
--- a/storage/myisammrg/ha_myisammrg.h
+++ b/storage/myisammrg/ha_myisammrg.h
@@ -35,6 +35,7 @@ class ha_myisammrg: public handler
   ulonglong table_flags() const
   {
     return (HA_REC_NOT_IN_SEQ | HA_AUTO_PART_KEY | HA_NO_TRANSACTIONS |
+            HA_BINLOG_ROW_CAPABLE | HA_BINLOG_STMT_CAPABLE |
 	    HA_NULL_IN_KEY | HA_CAN_INDEX_BLOBS | HA_FILE_BASED |
             HA_ANY_INDEX_MAY_BE_UNIQUE | HA_CAN_BIT_FIELD |
             HA_NO_COPY_ON_ALTER);