--source include/have_ndb.inc
--source include/have_binlog_format_row.inc
--source include/master-slave.inc

#
# Currently test only works with ndb since it retrieves "old"
# binlog positions with cluster_replication.binlog_index and apply_status;
#

# stop the save
connection slave;
STOP SLAVE;
CREATE DATABASE ndbsynctest;
USE ndbsynctest;

# get some data on the master
connection master;
CREATE DATABASE ndbsynctest;
USE ndbsynctest;
CREATE TABLE t1 (c1 BIT(1) NOT NULL, c2 BIT(1) NOT NULL, c3 CHAR(15), PRIMARY KEY(c3)) ENGINE = NDB ;
INSERT INTO t1 VALUES (1,1,"row1"),(0,1,"row2"),(1,0,"row3"),(0,0,"row4");
CREATE TABLE t2 (c1 CHAR(15), c2 BIT(1) NOT NULL, c3 BIT(1) NOT NULL, PRIMARY KEY(c1)) ENGINE = NDB ;
INSERT INTO t2 VALUES ("ABC",1,1),("BCDEF",0,1),("CD",1,0),("DEFGHIJKL",0,0);
SELECT hex(c1),hex(c2),c3 FROM t1 ORDER BY c3;
SELECT hex(c2),hex(c3),c1 FROM t2 ORDER BY c1;

# take a backup on master
--exec $NDB_MGM --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT" -e "start backup" >> $NDB_TOOLS_OUTPUT
--exec $NDB_TOOLS_DIR/ndb_select_all --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT" -d sys --delimiter=',' SYSTAB_0 | grep 520093696 > var/tmp.dat
CREATE TEMPORARY TABLE IF NOT EXISTS cluster_replication.backup_info (id INT, backup_id INT)ENGINE=HEAP;
DELETE FROM cluster_replication.backup_info;
LOAD DATA INFILE '../../var/tmp.dat' INTO TABLE cluster_replication.backup_info FIELDS TERMINATED BY ',';
--replace_column 1 <the_backup_id>
SELECT @the_backup_id:=backup_id FROM cluster_replication.backup_info;
let the_backup_id=`select @the_backup_id` ;
DROP TABLE cluster_replication.backup_info;
# update a row
UPDATE t1 SET c2=0 WHERE c3="row2";
SELECT hex(c1),hex(c2),c3 FROM t1 ORDER BY c3;

# restore on slave, first check that nothing is there
connection slave

# we should have no tables
SHOW TABLES;

--exec $NDB_TOOLS_DIR/ndb_restore --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT_SLAVE" -b $the_backup_id -n 1 -m -r --print --print_meta $NDB_BACKUP_DIR/BACKUP/BACKUP-$the_backup_id >> $NDB_TOOLS_OUTPUT
--exec $NDB_TOOLS_DIR/ndb_restore --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT_SLAVE" -b $the_backup_id -n 2 -r -e --print --print_meta $NDB_BACKUP_DIR/BACKUP/BACKUP-$the_backup_id >> $NDB_TOOLS_OUTPUT

#
# BUG#11960
# prior to bugfix "DROP DATABASE" would give a warning since
# the events were not created by ndb_restore
#
DROP DATABASE ndbsynctest;
CREATE DATABASE ndbsynctest;
USE ndbsynctest;
--exec $NDB_TOOLS_DIR/ndb_restore --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT_SLAVE" -b $the_backup_id -n 1 -m -r --print --print_meta $NDB_BACKUP_DIR/BACKUP/BACKUP-$the_backup_id >> $NDB_TOOLS_OUTPUT
--exec $NDB_TOOLS_DIR/ndb_restore --no-defaults --ndb-connectstring="localhost:$NDBCLUSTER_PORT_SLAVE" -b $the_backup_id -n 2 -r -e --print --print_meta $NDB_BACKUP_DIR/BACKUP/BACKUP-$the_backup_id >> $NDB_TOOLS_OUTPUT

# continue test
SHOW TABLES;

SELECT hex(c1),hex(c2),c3 FROM t1 ORDER BY c3;
SELECT hex(c2),hex(c3),c1 FROM t2 ORDER BY c1;

#
# now setup replication to continue from last epoch
# 1. get apply_status epoch from slave
# 2. get corresponding _next_ binlog postition from master
# 3. change master on slave

# 1.
connection slave;
--replace_column 1 <the_epoch>
SELECT @the_epoch:=MAX(epoch) FROM cluster_replication.apply_status;
let $the_epoch= `select @the_epoch` ;

# 2.
connection master;
--replace_result $the_epoch <the_epoch>
--replace_column 1 <the_pos>
eval SELECT @the_pos:=Position,@the_file:=SUBSTRING_INDEX(FILE, '/', -1)
   FROM cluster_replication.binlog_index WHERE epoch > $the_epoch ORDER BY epoch ASC LIMIT 1;
let $the_pos= `SELECT @the_pos` ;
let $the_file= `SELECT @the_file` ;

# 3.
connection slave;
--replace_result $the_pos <the_pos>
eval CHANGE MASTER TO
  master_log_file = '$the_file',
  master_log_pos = $the_pos ;
START SLAVE;

#
#
#
connection master;
#sync_slave_with_master;
--sleep 2
connection slave;
--replace_result $MASTER_MYPORT MASTER_PORT
--replace_column 1 <Slave_IO_State> 7 <Read_Master_Log_Pos> 8 <Relay_Log_File> 9 <Relay_Log_Pos> 16 <Replicate_Ignore_Table> 22 <Exec_Master_Log_Pos> 23 <Relay_Log_Space> 33 <Seconds_Behind_Master>
SHOW SLAVE STATUS;

SELECT hex(c1),hex(c2),c3 FROM t1 ORDER BY c3;
SELECT hex(c2),hex(c3),c1 FROM t2 ORDER BY c1;

#
# Cleanup
#

connection master;
DROP DATABASE ndbsynctest;
#sync_slave_with_master;
--sleep 2
connection slave;
STOP SLAVE;

#
# Test some replication commands
#
connection master;
reset master;
# should now contain nothing
select * from cluster_replication.binlog_index;

connection slave;
reset slave;
# should now contain nothing
select * from cluster_replication.apply_status;

# End 5.1 Test