Bug #27076 Cluster does not honor SQL_LOG_BIN flag

- both for data schema operations
- also make sure schema events vet the right server id when injected into the binlog
- use same mechanism to signal server_id in bug#17095, and reserve some "id's" for flagging special conditions on the event, in this case do not log it
- enable printing of server ids in the testcases to show that we cot it right

mysql-test/include/show_binlog_events.inc

 --let $binlog_start=106
 --replace_result $binlog_start <binlog_start>
---replace_column 2 # 4 # 5 #
+--replace_column 2 # 5 #
 --replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/
 --eval show binlog events from $binlog_start

mysql-test/r/ndb_binlog_ddl_multi.result

@@ -10,30 +10,30 @@ create table t1 (a int primary key) engine=ndb;
 create table t2 (a int primary key) engine=ndb;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	create database mysqltest
-master-bin1.000001	#	Query	#	#	use `mysqltest`; create table t1 (a int primary key) engine=ndb
-master-bin1.000001	#	Query	#	#	use `test`; create table t2 (a int primary key) engine=ndb
+master-bin1.000001	#	Query	1	#	create database mysqltest
+master-bin1.000001	#	Query	1	#	use `mysqltest`; create table t1 (a int primary key) engine=ndb
+master-bin1.000001	#	Query	102	#	use `test`; create table t2 (a int primary key) engine=ndb
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	create database mysqltest
-master-bin.000001	#	Query	#	#	use `mysqltest`; create table t1 (a int primary key) engine=ndb
-master-bin.000001	#	Query	#	#	use `test`; create table t2 (a int primary key) engine=ndb
+master-bin.000001	#	Query	1	#	create database mysqltest
+master-bin.000001	#	Query	1	#	use `mysqltest`; create table t1 (a int primary key) engine=ndb
+master-bin.000001	#	Query	102	#	use `test`; create table t2 (a int primary key) engine=ndb
 reset master;
 reset master;
 alter table t2 add column (b int);
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	use `test`; alter table t2 add column (b int)
+master-bin.000001	#	Query	102	#	use `test`; alter table t2 add column (b int)
 reset master;
 reset master;
 ALTER DATABASE mysqltest CHARACTER SET latin1;
 drop table mysqltest.t1;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	ALTER DATABASE mysqltest CHARACTER SET latin1
+master-bin.000001	#	Query	102	#	ALTER DATABASE mysqltest CHARACTER SET latin1
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Query	#	#	ALTER DATABASE mysqltest CHARACTER SET latin1
+master-bin.000001	#	Query	102	#	ALTER DATABASE mysqltest CHARACTER SET latin1
 reset master;
 reset master;
 use test;
@@ -42,14 +42,14 @@ drop database mysqltest;
 create table t1 (a int primary key) engine=ndb;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	BEGIN
-master-bin1.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin1.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin1.000001	#	Query	#	#	COMMIT
-master-bin1.000001	#	Query	#	#	drop database mysqltest
-master-bin1.000001	#	Query	#	#	use `test`; create table t1 (a int primary key) engine=ndb
+master-bin1.000001	#	Query	102	#	BEGIN
+master-bin1.000001	#	Table_map	102	#	table_id: # (test.t2)
+master-bin1.000001	#	Table_map	102	#	table_id: # (mysql.ndb_apply_status)
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: # flags: STMT_END_F
+master-bin1.000001	#	Query	102	#	COMMIT
+master-bin1.000001	#	Query	1	#	drop database mysqltest
+master-bin1.000001	#	Query	1	#	use `test`; create table t1 (a int primary key) engine=ndb
 drop table t2;
 reset master;
 reset master;
@@ -83,63 +83,63 @@ DROP LOGFILE GROUP lg1
 ENGINE =NDB;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	CREATE LOGFILE GROUP lg1
+master-bin1.000001	#	Query	1	#	CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
 INITIAL_SIZE 16M
 UNDO_BUFFER_SIZE = 1M
 ENGINE=NDB
-master-bin1.000001	#	Query	#	#	ALTER LOGFILE GROUP lg1
+master-bin1.000001	#	Query	1	#	ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
 INITIAL_SIZE = 4M 
 ENGINE=NDB
-master-bin1.000001	#	Query	#	#	CREATE TABLESPACE ts1
+master-bin1.000001	#	Query	1	#	CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
 INITIAL_SIZE 12M
 ENGINE NDB
-master-bin1.000001	#	Query	#	#	ALTER TABLESPACE ts1
+master-bin1.000001	#	Query	1	#	ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile02.dat'
 INITIAL_SIZE = 4M 
 ENGINE=NDB
-master-bin1.000001	#	Query	#	#	ALTER TABLESPACE ts1 
+master-bin1.000001	#	Query	1	#	ALTER TABLESPACE ts1 
 DROP DATAFILE 'datafile.dat' 
 ENGINE = NDB
-master-bin1.000001	#	Query	#	#	ALTER TABLESPACE ts1 
+master-bin1.000001	#	Query	1	#	ALTER TABLESPACE ts1 
 DROP DATAFILE 'datafile02.dat' 
 ENGINE = NDB
-master-bin1.000001	#	Query	#	#	DROP TABLESPACE ts1 
+master-bin1.000001	#	Query	1	#	DROP TABLESPACE ts1 
 ENGINE = NDB
-master-bin1.000001	#	Query	#	#	DROP LOGFILE GROUP lg1 
+master-bin1.000001	#	Query	1	#	DROP LOGFILE GROUP lg1 
 ENGINE =NDB
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	CREATE LOGFILE GROUP lg1
+master-bin1.000001	#	Query	1	#	CREATE LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile.dat'
 INITIAL_SIZE 16M
 UNDO_BUFFER_SIZE = 1M
 ENGINE=NDB
-master-bin1.000001	#	Query	#	#	ALTER LOGFILE GROUP lg1
+master-bin1.000001	#	Query	1	#	ALTER LOGFILE GROUP lg1
 ADD UNDOFILE 'undofile02.dat'
 INITIAL_SIZE = 4M 
 ENGINE=NDB
-master-bin1.000001	#	Query	#	#	CREATE TABLESPACE ts1
+master-bin1.000001	#	Query	1	#	CREATE TABLESPACE ts1
 ADD DATAFILE 'datafile.dat'
 USE LOGFILE GROUP lg1
 INITIAL_SIZE 12M
 ENGINE NDB
-master-bin1.000001	#	Query	#	#	ALTER TABLESPACE ts1
+master-bin1.000001	#	Query	1	#	ALTER TABLESPACE ts1
 ADD DATAFILE 'datafile02.dat'
 INITIAL_SIZE = 4M 
 ENGINE=NDB
-master-bin1.000001	#	Query	#	#	ALTER TABLESPACE ts1 
+master-bin1.000001	#	Query	1	#	ALTER TABLESPACE ts1 
 DROP DATAFILE 'datafile.dat' 
 ENGINE = NDB
-master-bin1.000001	#	Query	#	#	ALTER TABLESPACE ts1 
+master-bin1.000001	#	Query	1	#	ALTER TABLESPACE ts1 
 DROP DATAFILE 'datafile02.dat' 
 ENGINE = NDB
-master-bin1.000001	#	Query	#	#	DROP TABLESPACE ts1 
+master-bin1.000001	#	Query	1	#	DROP TABLESPACE ts1 
 ENGINE = NDB
-master-bin1.000001	#	Query	#	#	DROP LOGFILE GROUP lg1 
+master-bin1.000001	#	Query	1	#	DROP LOGFILE GROUP lg1 
 ENGINE =NDB
 drop table t1;
 reset master;
@@ -154,13 +154,13 @@ create table t3 (a int key) engine=ndb;
 rename table t3 to t4, t2 to t3, t1 to t2, t4 to t1;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	use `test`; create table t1 (a int key) engine=ndb
-master-bin1.000001	#	Query	#	#	use `test`; create table t2 (a int key) engine=ndb
-master-bin1.000001	#	Query	#	#	use `test`; create table t3 (a int key) engine=ndb
-master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t3` to `test.t4`
-master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t2` to `test.t3`
-master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t1` to `test.t2`
-master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t4` to `test.t1`
+master-bin1.000001	#	Query	1	#	use `test`; create table t1 (a int key) engine=ndb
+master-bin1.000001	#	Query	1	#	use `test`; create table t2 (a int key) engine=ndb
+master-bin1.000001	#	Query	1	#	use `test`; create table t3 (a int key) engine=ndb
+master-bin1.000001	#	Query	1	#	use `test`; rename table `test.t3` to `test.t4`
+master-bin1.000001	#	Query	1	#	use `test`; rename table `test.t2` to `test.t3`
+master-bin1.000001	#	Query	1	#	use `test`; rename table `test.t1` to `test.t2`
+master-bin1.000001	#	Query	1	#	use `test`; rename table `test.t4` to `test.t1`
 drop table t1;
 drop table t2;
 drop table t3;
@@ -177,18 +177,18 @@ insert into t2 values(2);
 drop table t2;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	use `test`; create table t1 (a int key) engine=ndb
-master-bin1.000001	#	Query	#	#	BEGIN
-master-bin1.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin1.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin1.000001	#	Query	#	#	COMMIT
-master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t1` to `test.t2`
-master-bin1.000001	#	Query	#	#	BEGIN
-master-bin1.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin1.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin1.000001	#	Query	#	#	COMMIT
-master-bin1.000001	#	Query	#	#	use `test`; drop table t2
+master-bin1.000001	#	Query	1	#	use `test`; create table t1 (a int key) engine=ndb
+master-bin1.000001	#	Query	102	#	BEGIN
+master-bin1.000001	#	Table_map	102	#	table_id: # (test.t1)
+master-bin1.000001	#	Table_map	102	#	table_id: # (mysql.ndb_apply_status)
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: # flags: STMT_END_F
+master-bin1.000001	#	Query	102	#	COMMIT
+master-bin1.000001	#	Query	1	#	use `test`; rename table `test.t1` to `test.t2`
+master-bin1.000001	#	Query	102	#	BEGIN
+master-bin1.000001	#	Table_map	102	#	table_id: # (test.t2)
+master-bin1.000001	#	Table_map	102	#	table_id: # (mysql.ndb_apply_status)
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: # flags: STMT_END_F
+master-bin1.000001	#	Query	102	#	COMMIT
+master-bin1.000001	#	Query	102	#	use `test`; drop table t2

mysql-test/r/ndb_binlog_discover.result

@@ -3,15 +3,15 @@ create table t1 (a int key) engine=ndb;
 reset master;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000001	#	Incident	#	#	#1 (LOST_EVENTS)
-master-bin.000001	#	Rotate	#	#	master-bin.000002;pos=4
+master-bin.000001	#	Incident	1	#	#1 (LOST_EVENTS)
+master-bin.000001	#	Rotate	1	#	master-bin.000002;pos=4
 PURGE MASTER LOGS TO 'master-bin.000002';
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin.000002	#	Query	#	#	BEGIN
-master-bin.000002	#	Table_map	#	#	table_id: # (test.t1)
-master-bin.000002	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin.000002	#	Write_rows	#	#	table_id: #
-master-bin.000002	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin.000002	#	Query	#	#	COMMIT
+master-bin.000002	#	Query	1	#	BEGIN
+master-bin.000002	#	Table_map	1	#	table_id: # (test.t1)
+master-bin.000002	#	Table_map	1	#	table_id: # (mysql.ndb_apply_status)
+master-bin.000002	#	Write_rows	1	#	table_id: #
+master-bin.000002	#	Write_rows	1	#	table_id: # flags: STMT_END_F
+master-bin.000002	#	Query	1	#	COMMIT
 drop table t1;

mysql-test/r/ndb_binlog_ignore_db.result

@@ -7,5 +7,5 @@ create table t1 (a int primary key, b int) engine=ndb;
 insert into t1 values (1, 1);
 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 if exists t1
+master-bin.000001	#	Query	1	#	use `test`; drop table if exists t1
 drop database mysqltest;

mysql-test/r/ndb_binlog_multi.result

@@ -8,14 +8,14 @@ CREATE TABLE t2 (a INT PRIMARY KEY, b int) ENGINE = NDB;
 INSERT INTO t2 VALUES (1,1),(2,2);
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	use `test`; CREATE TABLE t2 (a INT PRIMARY KEY, b int) ENGINE = NDB
-master-bin1.000001	#	Query	#	#	BEGIN
-master-bin1.000001	#	Table_map	#	#	table_id: # (test.t2)
-master-bin1.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	102	#	use `test`; CREATE TABLE t2 (a INT PRIMARY KEY, b int) ENGINE = NDB
+master-bin1.000001	#	Query	102	#	BEGIN
+master-bin1.000001	#	Table_map	102	#	table_id: # (test.t2)
+master-bin1.000001	#	Table_map	102	#	table_id: # (mysql.ndb_apply_status)
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: # flags: STMT_END_F
+master-bin1.000001	#	Query	102	#	COMMIT
 select * from t2 order by a;
 a	b
 1	1
@@ -31,15 +31,15 @@ a	b
 DROP TABLE t2;
 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 t2 (a INT PRIMARY KEY, b int) ENGINE = NDB
-master-bin.000001	#	Query	#	#	BEGIN
-master-bin.000001	#	Table_map	#	#	table_id: # (test.t2)
-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: # flags: STMT_END_F
-master-bin.000001	#	Query	#	#	COMMIT
-master-bin.000001	#	Query	#	#	use `test`; DROP TABLE t2
+master-bin.000001	#	Query	102	#	use `test`; CREATE TABLE t2 (a INT PRIMARY KEY, b int) ENGINE = NDB
+master-bin.000001	#	Query	1	#	BEGIN
+master-bin.000001	#	Table_map	1	#	table_id: # (test.t2)
+master-bin.000001	#	Table_map	1	#	table_id: # (mysql.ndb_apply_status)
+master-bin.000001	#	Write_rows	1	#	table_id: #
+master-bin.000001	#	Write_rows	1	#	table_id: #
+master-bin.000001	#	Write_rows	1	#	table_id: # flags: STMT_END_F
+master-bin.000001	#	Query	1	#	COMMIT
+master-bin.000001	#	Query	1	#	use `test`; DROP TABLE t2
 SELECT inserts,updates,deletes,schemaops FROM 
 mysql.ndb_binlog_index WHERE epoch=<the_epoch>;
 inserts	updates	deletes	schemaops
@@ -50,14 +50,14 @@ CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = NDB;
 INSERT INTO t1 VALUES (1),(2);
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = NDB
-master-bin1.000001	#	Query	#	#	BEGIN
-master-bin1.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin1.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	102	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = NDB
+master-bin1.000001	#	Query	102	#	BEGIN
+master-bin1.000001	#	Table_map	102	#	table_id: # (test.t1)
+master-bin1.000001	#	Table_map	102	#	table_id: # (mysql.ndb_apply_status)
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: # flags: STMT_END_F
+master-bin1.000001	#	Query	102	#	COMMIT
 SELECT @the_epoch2:=epoch,inserts,updates,deletes,schemaops FROM 
 mysql.ndb_binlog_index ORDER BY epoch DESC LIMIT 1;
 @the_epoch2:=epoch	inserts	updates	deletes	schemaops
@@ -69,15 +69,15 @@ inserts	updates	deletes	schemaops
 drop table t1;
 show binlog events from <binlog_start>;
 Log_name	Pos	Event_type	Server_id	End_log_pos	Info
-master-bin1.000001	#	Query	#	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = NDB
-master-bin1.000001	#	Query	#	#	BEGIN
-master-bin1.000001	#	Table_map	#	#	table_id: # (test.t1)
-master-bin1.000001	#	Table_map	#	#	table_id: # (mysql.ndb_apply_status)
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: #
-master-bin1.000001	#	Write_rows	#	#	table_id: # flags: STMT_END_F
-master-bin1.000001	#	Query	#	#	COMMIT
-master-bin1.000001	#	Query	#	#	use `test`; drop table t1
+master-bin1.000001	#	Query	102	#	use `test`; CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE = NDB
+master-bin1.000001	#	Query	102	#	BEGIN
+master-bin1.000001	#	Table_map	102	#	table_id: # (test.t1)
+master-bin1.000001	#	Table_map	102	#	table_id: # (mysql.ndb_apply_status)
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: #
+master-bin1.000001	#	Write_rows	102	#	table_id: # flags: STMT_END_F
+master-bin1.000001	#	Query	102	#	COMMIT
+master-bin1.000001	#	Query	102	#	use `test`; drop table t1
 SELECT inserts,updates,deletes,schemaops FROM
 mysql.ndb_binlog_index WHERE epoch > <the_epoch> AND epoch <= <the_epoch2>;
 inserts	updates	deletes	schemaops

sql/ha_ndbcluster.cc

@@ -350,7 +350,7 @@ Thd_ndb::Thd_ndb()
   count= 0;
   all= NULL;
   stmt= NULL;
-  error= 0;
+  m_error= FALSE;
   query_state&= NDB_QUERY_NORMAL;
   options= 0;
   (void) hash_init(&open_tables, &my_charset_bin, 5, 0, 0,
@@ -385,7 +385,7 @@ void
 Thd_ndb::init_open_tables()
 {
   count= 0;
-  error= 0;
+  m_error= FALSE;
   my_hash_reset(&open_tables);
 }
 
@@ -461,7 +461,7 @@ ha_rows ha_ndbcluster::records()
   }
 
   THD *thd= current_thd;
-  if (get_thd_ndb(thd)->error)
+  if (get_thd_ndb(thd)->m_error)
     local_info->no_uncommitted_rows_count= 0;
 
   DBUG_RETURN(retval + local_info->no_uncommitted_rows_count);
@@ -492,7 +492,7 @@ int ha_ndbcluster::records_update()
   }
   {
     THD *thd= current_thd;
-    if (get_thd_ndb(thd)->error)
+    if (get_thd_ndb(thd)->m_error)
       local_info->no_uncommitted_rows_count= 0;
   }
   if (result == 0)
@@ -505,7 +505,7 @@ void ha_ndbcluster::no_uncommitted_rows_execute_failure()
   if (m_ha_not_exact_count)
     return;
   DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_execute_failure");
-  get_thd_ndb(current_thd)->error= 1;
+  get_thd_ndb(current_thd)->m_error= TRUE;
   DBUG_VOID_RETURN;
 }
 
@@ -529,7 +529,7 @@ void ha_ndbcluster::no_uncommitted_rows_reset(THD *thd)
   DBUG_ENTER("ha_ndbcluster::no_uncommitted_rows_reset");
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
   thd_ndb->count++;
-  thd_ndb->error= 0;
+  thd_ndb->m_error= FALSE;
   DBUG_VOID_RETURN;
 }
 
@@ -2713,9 +2713,13 @@ int ha_ndbcluster::write_row(byte *record)
     op->setValue(no_fields, part_func_value);
   }
 
-  if (thd->slave_thread)
+  if (unlikely(m_slow_path))
+  {
+    if (!(thd->options & OPTION_BIN_LOG))
+      op->setAnyValue(NDB_ANYVALUE_FOR_NOLOGGING);
+    else if (thd->slave_thread)
       op->setAnyValue(thd->server_id);
-
+  }
   m_rows_changed++;
 
   /*
@@ -2997,9 +3001,13 @@ int ha_ndbcluster::update_row(const byte *old_data, byte *new_data)
     op->setValue(no_fields, part_func_value);
   }
 
-  if (thd->slave_thread)
+  if (unlikely(m_slow_path))
+  {
+    if (!(thd->options & OPTION_BIN_LOG))
+      op->setAnyValue(NDB_ANYVALUE_FOR_NOLOGGING);
+    else if (thd->slave_thread)
       op->setAnyValue(thd->server_id);
-
+  }
   /*
     Execute update operation if we are not doing a scan for update
     and there exist UPDATE AFTER triggers
@@ -3060,9 +3068,15 @@ int ha_ndbcluster::delete_row(const byte *record)
 
     no_uncommitted_rows_update(-1);
 
-    if (thd->slave_thread)
-      ((NdbOperation *)trans->getLastDefinedOperation())->setAnyValue(thd->server_id);
-
+    if (unlikely(m_slow_path))
+    {
+      if (!(thd->options & OPTION_BIN_LOG))
+        ((NdbOperation *)trans->getLastDefinedOperation())->
+          setAnyValue(NDB_ANYVALUE_FOR_NOLOGGING);
+      else if (thd->slave_thread)
+        ((NdbOperation *)trans->getLastDefinedOperation())->
+          setAnyValue(thd->server_id);
+    }
     if (!(m_primary_key_update || m_delete_cannot_batch))
       // If deleting from cursor, NoCommit will be handled in next_result
       DBUG_RETURN(0);
@@ -3093,9 +3107,14 @@ int ha_ndbcluster::delete_row(const byte *record)
         DBUG_RETURN(error);
     }
 
-    if (thd->slave_thread)
+    if (unlikely(m_slow_path))
+    {
+      if (!(thd->options & OPTION_BIN_LOG))
+        op->setAnyValue(NDB_ANYVALUE_FOR_NOLOGGING);
+      else if (thd->slave_thread)
         op->setAnyValue(thd->server_id);
     }
+  }
 
   // Execute delete operation
   if (execute_no_commit(this,trans,FALSE) != 0) {
@@ -4268,6 +4287,10 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
         thd_ndb->stmt= trans;
 	thd_ndb->query_state&= NDB_QUERY_NORMAL;
         thd_ndb->trans_options= 0;
+        thd_ndb->m_slow_path= FALSE;
+        if (thd->slave_thread ||
+            !(thd->options & OPTION_BIN_LOG))
+          thd_ndb->m_slow_path= TRUE;
         trans_register_ha(thd, FALSE, ndbcluster_hton);
       } 
       else 
@@ -4285,6 +4308,10 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
           thd_ndb->all= trans; 
 	  thd_ndb->query_state&= NDB_QUERY_NORMAL;
           thd_ndb->trans_options= 0;
+          thd_ndb->m_slow_path= FALSE;
+          if (thd->slave_thread ||
+              !(thd->options & OPTION_BIN_LOG))
+            thd_ndb->m_slow_path= TRUE;
           trans_register_ha(thd, TRUE, ndbcluster_hton);
 
           /*
@@ -4325,9 +4352,13 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
     // Start of transaction
     m_rows_changed= 0;
     m_ops_pending= 0;
+    m_slow_path= thd_ndb->m_slow_path;
 #ifdef HAVE_NDB_BINLOG
+    if (unlikely(m_slow_path))
+    {
       if (m_share == ndb_apply_status_share && thd->slave_thread)
         thd_ndb->trans_options|= TNTO_INJECTED_APPLY_STATUS;
+    }
 #endif
     // TODO remove double pointers...
     m_thd_ndb_share= thd_ndb->get_open_table(thd, m_table);
@@ -4474,8 +4505,12 @@ static int ndbcluster_commit(handlerton *hton, THD *thd, bool all)
     DBUG_RETURN(0);
 
 #ifdef HAVE_NDB_BINLOG
+  if (unlikely(thd_ndb->m_slow_path))
+  {
     if (thd->slave_thread)
-    ndbcluster_update_apply_status(thd, thd_ndb->trans_options & TNTO_INJECTED_APPLY_STATUS);
+      ndbcluster_update_apply_status
+        (thd, thd_ndb->trans_options & TNTO_INJECTED_APPLY_STATUS);
+  }
 #endif /* HAVE_NDB_BINLOG */
 
   if (execute_commit(thd,trans) != 0)

sql/ha_ndbcluster.h

@@ -623,7 +623,8 @@ class Thd_ndb
   uint lock_count;
   NdbTransaction *all;
   NdbTransaction *stmt;
-  int error;
+  bool m_error;
+  bool m_slow_path;
   uint32 options;
   uint32 trans_options;
   List<NDB_SHARE> changed_tables;
@@ -969,6 +970,7 @@ private:
   ha_rows m_ops_pending;
   bool m_skip_auto_increment;
   bool m_blobs_pending;
+  bool m_slow_path;
   my_ptrdiff_t m_blobs_offset;
   // memory for blobs in one tuple
   char *m_blobs_buffer;

sql/ha_ndbcluster_binlog.cc

@@ -872,6 +872,7 @@ struct Cluster_schema
   uint32 id;
   uint32 version;
   uint32 type;
+  uint32 any_value;
 };
 
 /*
@@ -1392,6 +1393,12 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
       /* type */
       r|= op->setValue(SCHEMA_TYPE_I, log_type);
       DBUG_ASSERT(r == 0);
+      /* any value */
+      if (!(thd->options & OPTION_BIN_LOG))
+        r|= op->setAnyValue(NDB_ANYVALUE_FOR_NOLOGGING);
+      else
+        r|= op->setAnyValue(thd->server_id);
+      DBUG_ASSERT(r == 0);
       if (log_db != new_db && new_db && new_table_name)
       {
         log_db= new_db;
@@ -1737,6 +1744,31 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
   DBUG_RETURN(0);
 }
 
+static void ndb_binlog_query(THD *thd, Cluster_schema *schema)
+{
+  if (schema->any_value & NDB_ANYVALUE_RESERVED)
+  {
+    if (schema->any_value != NDB_ANYVALUE_FOR_NOLOGGING)
+      sql_print_warning("NDB: unknown value for binlog signalling 0x%X, "
+                        "query not logged",
+                        schema->any_value);
+    return;
+  }
+  uint32 thd_server_id_save= thd->server_id;
+  DBUG_ASSERT(sizeof(thd_server_id_save) == sizeof(thd->server_id));
+  char *thd_db_save= thd->db;
+  if (schema->any_value == 0)
+    thd->server_id= ::server_id;
+  else
+    thd->server_id= schema->any_value;
+  thd->db= schema->db;
+  thd->binlog_query(THD::STMT_QUERY_TYPE, schema->query,
+                    schema->query_length, FALSE,
+                    schema->name[0] == 0 || thd->db[0] == 0);
+  thd->server_id= thd_server_id_save;
+  thd->db= thd_db_save;
+}
+
 static int
 ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
                                       NdbEventOperation *pOp,
@@ -1761,7 +1793,10 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
       MY_BITMAP slock;
       bitmap_init(&slock, schema->slock, 8*SCHEMA_SLOCK_SIZE, FALSE);
       uint node_id= g_ndb_cluster_connection->node_id();
+      {
         ndbcluster_get_schema(tmp_share, schema);
+        schema->any_value= pOp->getAnyValue();
+      }
       enum SCHEMA_OP_TYPE schema_type= (enum SCHEMA_OP_TYPE)schema->type;
       DBUG_PRINT("info",
                  ("%s.%s: log query_length: %d  query: '%s'  type: %d",
@@ -1885,7 +1920,8 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           run_query(thd, schema->query,
                     schema->query + schema->query_length,
                     TRUE,    /* print error */
-                    FALSE);  /* binlog the query */
+                    TRUE);   /* don't binlog the query */
+          log_query= 1;
           break;
         case SOT_TABLESPACE:
         case SOT_LOGFILE_GROUP:
@@ -1895,14 +1931,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           abort();
         }
         if (log_query && ndb_binlog_running)
-        {
-          char *thd_db_save= thd->db;
-          thd->db= schema->db;
-          thd->binlog_query(THD::STMT_QUERY_TYPE, schema->query,
-                            schema->query_length, FALSE,
-                            schema->name[0] == 0 || thd->db[0] == 0);
-          thd->db= thd_db_save;
-        }
+          ndb_binlog_query(thd, schema);
         /* signal that schema operation has been handled */
         DBUG_DUMP("slock", (char*)schema->slock, schema->slock_length);
         if (bitmap_is_set(&slock, node_id))
@@ -2160,14 +2189,7 @@ ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
       }
     }
     if (ndb_binlog_running && log_query)
-    {
-      char *thd_db_save= thd->db;
-      thd->db= schema->db;
-      thd->binlog_query(THD::STMT_QUERY_TYPE, schema->query,
-                        schema->query_length, FALSE,
-                        schema->name[0] == 0);
-      thd->db= thd_db_save;
-    }
+      ndb_binlog_query(thd, schema);
   }
   while ((schema= post_epoch_unlock_list->pop()))
   {
@@ -2323,6 +2345,18 @@ int ndbcluster_binlog_start()
 {
   DBUG_ENTER("ndbcluster_binlog_start");
 
+  if (::server_id == 0)
+  {
+    sql_print_warning("NDB: server id set to zero will cause any other mysqld "
+                      "with bin log to log with wrong server id");
+  }
+  else if (::server_id & 0x1 << 31)
+  {
+    sql_print_error("NDB: server id's with high bit set is reserved for internal "
+                    "purposes");
+    DBUG_RETURN(-1);
+  }
+
   pthread_mutex_init(&injector_mutex, MY_MUTEX_INIT_FAST);
   pthread_cond_init(&injector_cond, NULL);
   pthread_mutex_init(&ndb_schema_share_mutex, MY_MUTEX_INIT_FAST);
@@ -3192,9 +3226,17 @@ ndb_binlog_thread_handle_data_event(Ndb *ndb, NdbEventOperation *pOp,
   if (share == ndb_apply_status_share)
     return 0;
 
-  uint originating_server_id= pOp->getAnyValue();
+  uint32 originating_server_id= pOp->getAnyValue();
   if (originating_server_id == 0)
     originating_server_id= ::server_id;
+  else if (originating_server_id & NDB_ANYVALUE_RESERVED)
+  {
+    if (originating_server_id != NDB_ANYVALUE_FOR_NOLOGGING)
+      sql_print_warning("NDB: unknown value for binlog signalling 0x%X, "
+                        "event not logged",
+                        originating_server_id);
+    return 0;
+  }
 
   TABLE *table= share->table;
   DBUG_ASSERT(trans.good());

sql/ha_ndbcluster_binlog.h

@@ -30,6 +30,10 @@ extern ulong ndb_extra_logging;
 
 #define NDB_INVALID_SCHEMA_OBJECT 241
 
+/* server id's with high bit set is reservered */
+#define NDB_ANYVALUE_FOR_NOLOGGING 0xFFFFFFFF
+#define NDB_ANYVALUE_RESERVED      0x80000000
+
 extern handlerton *ndbcluster_hton;
 
 /*