########################################################
# Test binlog variants produced by Ndb
#
#  1) Updates logged as write_row events
#     Only primary key and updated columns included in the 
#     event
#  2) Updates logged as write_row_events
#     All columns included in the event
#  3) Updates logged as update_row events
#     Only primary key and updated columns included in the
#     event
#  4) Updates logged as update_row events
#     All columns included in the event
#
#  Format variant (1) is the Ndb default.
#
#  We use mysqlbinlog --verbose to check that the
#  generated binlog contents are as expected.
#
########################################################
-- source include/have_ndb.inc
-- source include/have_binlog_format_row.inc

# Setup connections
connect(mysqld1,127.0.0.1,root,,test,$MASTER_MYPORT);
connect(mysqld2,127.0.0.1,root,,test,$MASTER_MYPORT1);
connect(mysqld3,127.0.0.1,root,,test,$MASTER_MYPORT2);
connect(mysqld4,127.0.0.1,root,,test,$MASTER_MYPORT3);
connect(mysqld5,127.0.0.1,root,,test,$MASTER_MYPORT4);
connect(mysqld6,127.0.0.1,root,,test,$MASTER_MYPORT5);

connection mysqld1;

# Create the tables we're going to use
create table ab(a1 int, a2 int, a3 int, a4 int, a5 int, primary key(a2,a5)) engine = ndb;
create table ba(ks int primary key, st int, lp int) engine = ndb;
create table cpk(nonchar int, charone varchar(100), chartwo varchar(100),
                 other int,
                 primary key (nonchar, charone, chartwo)) engine=ndb;

# Wait for each mysqld to startup binlogging
--let $source_server=mysqld1
--let $dest_server=mysqld2
source suite/ndb_rpl/t/wait_schema_logging.inc;

--let $source_server=mysqld1
--let $dest_server=mysqld3
source suite/ndb_rpl/t/wait_schema_logging.inc;

--let $source_server=mysqld1
--let $dest_server=mysqld4
source suite/ndb_rpl/t/wait_schema_logging.inc;

--let $source_server=mysqld1
--let $dest_server=mysqld5
source suite/ndb_rpl/t/wait_schema_logging.inc;

--let $source_server=mysqld1
--let $dest_server=mysqld6
source suite/ndb_rpl/t/wait_schema_logging.inc;

--disable_query_log
connection mysqld1;
reset master;
connection mysqld2;
reset master;
connection mysqld3;
reset master;
connection mysqld4;
reset master;
connection mysqld5;
reset master;
connection mysqld6;
reset master;
--enable_query_log

connection mysqld1;

# Now make the inserts and update
insert into ab values (1,1,1,1,1), (2,2,2,2,2), (3,3,3,3,3), (4,4,4,4,4);
update ab set a3=111 where a2=1;
update ab set a1=222, a4=222 where a2=2;

insert into ba values (1, 1, 1), (2,2,2), (3,3,3), (4,4,4);
update ba set lp=40 where ks=4; # Update does not affect all columns
delete from ba where ks=2;      # Also a delete for fun

insert into cpk values(1, "ABCD", "WXYZ", 100);
insert into cpk values(2, "ABCD", "WXYZ", 100);
insert into cpk values(3, "ABCD", "WXYZ", 100);
insert into cpk values(4, "ABCD", "WXYZ", 100);
insert into cpk values(5, "ABCD", "WXYZ", 100);
insert into cpk values(6, "ABCD", "WXYZ", 100);

--echo Change one char key
update cpk set charone="abcd" where charone="ABCD" and nonchar=1 and chartwo="WXYZ";
--echo Change other char key
update cpk set chartwo="wxyz" where charone="ABCD" and nonchar=2 and chartwo="WXYZ";
--echo Change both char keys
update cpk set charone="abcd", chartwo="wxyz" where charone="ABCD" and nonchar=3 and chartwo="WXYZ";
--echo Null change
update cpk set charone="ABCD", chartwo="WXYZ" where charone="ABCD" and nonchar=4 and chartwo="WXYZ";
--echo Null change + data change
update cpk set charone="ABCD", chartwo="WXYZ", other=200 where charone="ABCD" and nonchar=5 and chartwo="WXYZ";
--echo Change both char keys + data
update cpk set charone="abcd", chartwo="wxyz", other=200 where charone="ABCD" and nonchar=6 and chartwo="WXYZ";



--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Now let's trim the Binlogs on each server

connection mysqld1;
--source include/wait_for_binlog_event.inc
flush logs;
connection mysqld2;
--source include/wait_for_binlog_event.inc
flush logs;
connection mysqld3;
--source include/wait_for_binlog_event.inc
flush logs;
connection mysqld4;
--source include/wait_for_binlog_event.inc
flush logs;
connection mysqld5;
--source include/wait_for_binlog_event.inc
flush logs;
connection mysqld6;
--source include/wait_for_binlog_event.inc
flush logs;

# Empty the table
delete from ba;

# Now let's examine the contents of the first binlog
# on each server
# We'll also apply the Binlog and check that the
# table contents are as expected in each case.
# As each server is recording in a new binlog, the
# new updates will go there.

connection mysqld1;

show variables like 'ndb_log_update%';
--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from ab order by a2,a4;
delete from ab;
select * from ba order by ks;
delete from ba;
select * from cpk order by nonchar;
delete from cpk;

connection mysqld2;

show variables like 'ndb_log_update%';
--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from ab order by a2,a4;
delete from ab;
select * from ba order by ks;
delete from ba;
select * from cpk order by nonchar;
delete from cpk;

connection mysqld3;

show variables like 'ndb_log_update%';
--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from ab order by a2,a4;
delete from ab;
select * from ba order by ks;
delete from ba;
select * from cpk order by nonchar;
delete from cpk;

connection mysqld4;

show variables like 'ndb_log_update%';
--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from ab order by a2,a4;
delete from ab;
select * from ba order by ks;
delete from ba;
select * from cpk order by nonchar;
delete from cpk;

connection mysqld5;

show variables like 'ndb_log_update%';
--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from ab order by a2,a4;
delete from ab;
select * from ba order by ks;
delete from ba;
select * from cpk order by nonchar;
delete from cpk;

connection mysqld6;

show variables like 'ndb_log_update%';
--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from ab order by a2,a4;
delete from ab;
select * from ba order by ks;
delete from ba;
select * from cpk order by nonchar;
delete from cpk;

drop table ab;
drop table ba;
drop table cpk;
--disable_query_log
flush logs;
--enable_query_log

connection mysqld1;
CREATE TABLE mysql.ndb_replication
  (db VARBINARY(63),
   table_name VARBINARY(63),
   server_id INT UNSIGNED,
   binlog_type INT UNSIGNED,
   conflict_fn VARBINARY(128),
   PRIMARY KEY USING HASH (db,table_name,server_id))
  ENGINE=NDB PARTITION BY KEY(db,table_name);

insert into mysql.ndb_replication values("test", "t1", 0, 6, NULL);
insert into mysql.ndb_replication values("test", "t2", 0, 7, NULL);
insert into mysql.ndb_replication values("test", "t3", 0, 8, NULL);
insert into mysql.ndb_replication values("test", "t4", 0, 9, NULL);

create table t1(a1 int, a2 int, a3 int, a4 int, a5 int, primary key(a2,a5)) engine=ndb;
create table t2(a1 int, a2 int, a3 int, a4 int, a5 int, primary key(a2,a5)) engine=ndb;
create table t3(a1 int, a2 int, a3 int, a4 int, a5 int, primary key(a2,a5)) engine=ndb;
create table t4(a1 int, a2 int, a3 int, a4 int, a5 int, primary key(a2,a5)) engine=ndb;

--disable_query_log
reset master;
--enable_query_log

# Now make the inserts and update
insert into t1 values (1,1,1,1,1), (2,2,2,2,2), (3,3,3,3,3), (4,4,4,4,4);
insert into t2 values (1,1,1,1,1), (2,2,2,2,2), (3,3,3,3,3), (4,4,4,4,4);
insert into t3 values (1,1,1,1,1), (2,2,2,2,2), (3,3,3,3,3), (4,4,4,4,4);
insert into t4 values (1,1,1,1,1), (2,2,2,2,2), (3,3,3,3,3), (4,4,4,4,4);
update t1 set a3=111 where a2=1;
update t1 set a1=222, a4=222 where a2=2;
update t2 set a3=111 where a2=1;
update t2 set a1=222, a4=222 where a2=2;
update t3 set a3=111 where a2=1;
update t3 set a1=222, a4=222 where a2=2;
update t4 set a3=111 where a2=1;
update t4 set a1=222, a4=222 where a2=2;

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Now let's trim the Binlogs on each server
connection mysqld1;
--source include/wait_for_binlog_event.inc
flush logs;

--source suite/ndb_binlog/t/ndb_binlog_get_binlog_stmts.inc

# Clear the table
delete from t1;
delete from t2;
delete from t3;
delete from t4;

--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from t1 order by a2,a5;
select * from t2 order by a2,a5;
select * from t3 order by a2,a5;
select * from t4 order by a2,a5;

drop table t1, t2, t3, t4;
drop table mysql.ndb_replication;

# Bug#46662
# Replicating changes to tables with unique indexes
# The fix to bug#27378 results in the slave using NdbApi's write()
# mechanism when applying WRITE_ROW events to tables with unique
# indices.
# 
# If this is not done then the slave attempts to partially use SQL 
# REPLACE semantics when applying WRITE_ROW events to tables with 
# unique indexes, which is not good and the slave fails with a 
# duplicate key error on the primary key.
#
# The fix to Bug#46662 aims to correct this, so that replicated
# updates to tables with unique indices can work.
# Note that other issues with replicating into tables with unique
# indexes remain.
# 

connection mysqld1;
reset master;
show variables like 'ndb_log_update%';

create table bah (tst int primary key, cvy int, sqs int, unique(sqs)) engine=ndb;

insert into bah values (1,1,1);

# Wait for epoch to complete in Binlog
--disable_query_log
create table dummy (a int primary key) engine=ndb;
--enable_query_log

# Now perform update
# This will be logged as WRITE
# Without ability to use NdbApi write() for replace, mysqlbinlog
# application will fail with duplicate key error on insert.
update bah set cvy= 2 where tst=1;

select * from bah order by tst;

# Wait for epoch to complete in Binlog
--disable_query_log
drop table dummy;
flush logs;
--enable_query_log

drop table bah;

# Now let's re-apply the binlog
# Without fix, this fails with duplicate PK error
--echo Manually applying captured binlog
--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000001 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from bah order by tst;

drop table bah;

# Bug #14615095   ERROR 839 'ILLEGAL NULL ATTRIBUTE' WHEN REPLAYING BINLOG
# When applying WRITE_ROW events to tables where the rows are missing
# any errors should be ignored

connection mysqld1;
reset master;
show variables like '%log_update%';

CREATE TABLE `t1` (
   `charId` varchar(60) NOT NULL,
   `enumId` enum('A','B','C') NOT NULL,
   `val` bigint(20) NOT NULL,
   `version` int(11) NOT NULL,
   PRIMARY KEY (`charId`,`enumId`)
  ) ENGINE=ndbcluster DEFAULT CHARSET=latin1;

INSERT INTO t1 VALUES ('', 'A', 0, 1);

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Wait until the INSERT statement is confirmed to have made it into the current binary log
--source include/wait_for_binlog_event.inc
FLUSH LOGS;

UPDATE t1 SET val = val + 1 WHERE charId = '';

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Wait until the UPDATE statement is confirmed to have made it into the current binary log
--let $wait_binlog_file=binlog.000002
--source include/wait_for_binlog_event.inc
FLUSH LOGS;

DELETE FROM t1 WHERE charId = '';

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Wait until the DELETE statement is confirmed to have made it into the current binary log
--let $wait_binlog_file=binlog.000003
--source include/wait_for_binlog_event.inc
FLUSH LOGS;

# Now let's re-apply the binlog from the UPDATE
# Without fix, this fails with 'Illegal null attribute'
--echo Manually applying captured binlog
--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000002 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
# Check that the table is still empty
select * from t1;

drop table t1;

# Bug #14678088   CAN'T FIND RECORD IN
# We need to be idempotent when applying binlog
# test insert of existing row, update and delete
# of non-existing row

connection mysqld1;
reset master;
show variables like '%log_update%';

create table t1 (pk int not null primary key, name varchar(256)) engine = ndb;

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Wait until all statements are confirmed to have made it into the current binary log
--let $wait_binlog_file=binlog.000001
--source include/wait_for_binlog_event.inc
FLUSH LOGS;

insert into t1 values (0, "zero"),(1,"one"),(2,"two"),(3,"three"),(4,"four"),(5,"five"),(6,"six"),(7,"seven"),(8,"eight"),(9,"nine");
select * from t1 order by pk;

update t1 set name = "even" where pk in (0,2,4,6,8);
update t1 set name = "odd" where pk in (1,3,5,7,9);

delete from t1 where name = "odd";

select * from t1 order by pk;

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Wait until all statements are confirmed to have made it into the current binary log
--let $wait_binlog_file=binlog.000002
--source include/wait_for_binlog_event.inc
FLUSH LOGS;

truncate t1;
insert into t1 values (0, "zero"),(1,"one"),(2,"two"),(3,"three"),(4,"four"),(5,"five"),(6,"six"),(7,"seven"),(8,"eight"),(9,"nine");
select * from t1 order by pk;

--disable_query_log
# Add an event-stream marker
create table stream_marker(a int) engine=ndb;
drop table stream_marker;
--let $wait_binlog_event=stream_marker
--enable_query_log

# Wait until all statements are confirmed to have made it into the current binary log
--let $wait_binlog_file=binlog.000003
--source include/wait_for_binlog_event.inc
FLUSH LOGS;

# Now let's re-apply the binlog INSERT,UPDATE,DELETE
# Without fix, this fails with 'Illegal null attribute'
--echo Manually applying captured binlog
--disable_query_log
let $MYSQLD_DATADIR= `select @@datadir;`;
--exec $MYSQL_BINLOG $MYSQLD_DATADIR/binlog.000002 > $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql
--exec $MYSQL -uroot < $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql

--enable_query_log
select * from t1 order by pk;

drop table t1;
--remove_file $MYSQLTEST_VARDIR/tmp/ndb_binlog_mysqlbinlog.sql