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);