#
# WL#1697: Multi source Replication
#
# This test verifies that the replication performance schema tables work fine
# with multi-source replication.
# General idea of the test:
# 1) test the count of the rows in all the tables are fine.
# 2) count decreases or increases with channels
# 3) Joins could be made without error
#
# Note: we do not test exclusively the correctness of each field of all
# channels. The tests to test the correctness of individual channel (i.e default)
# are present in rpl_perfschema_* tests.

#Skip on group replication runs
--source include/not_group_replication_plugin.inc
--source include/have_binlog_format_mixed.inc
--source include/not_have_privilege_checks_user.inc
# Test requires master-info-repository=TABLE, relay-log-info-repository=TABLE
--source include/have_slave_repository_type_table.inc
--echo #
--echo # Set up masters server_1 and server_3 with server_2 being a slave.
--echo #
--let $rpl_topology= 1->2,3->2
--let $rpl_multi_source= 1
--source include/rpl_init.inc

--echo #
--echo # Test case 1: Create multiple sources and make sure that
--echo # number of sources= number of rows in all the performance schema tables.
--echo # On master, where replication is not setup, it shall show 0 rows.
--let $rpl_connection_name= server_1
--source include/rpl_connection.inc

--let $num= `select count(*) from performance_schema.replication_connection_configuration`
--let $assert_text= Number of rows = 0 on master for repliation_connection_configuration table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_connection_status`
--let $assert_text= Number of rows = 0 on master for repliation_connection_status table.
--let $assert_cond= $num= 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_configuration`
--let $assert_text= Number of rows = 0 on master for repliation_applier_configuration table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_status`
--let $assert_text= Number of rows = 0 on master for repliation_applier_status table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_status_by_coordinator`
--let $assert_text= Number of rows = 0 on master for replication_applier_status_by coordinator table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_status_by_worker`
--let $assert_text= Number of rows = 0 on master for replication_applier_status_by_worker table.
--let $assert_cond= "$num" = 0;
--source include/assert.inc

--echo #
--echo # Test case 2: On the slave,  number of rows should be zero for a non existent channel
--echo #
--let $rpl_connection_name= server_2
--source include/rpl_connection.inc

--let $num= `select count(*) from performance_schema.replication_connection_configuration where channel_name="non_existent"`
--let $assert_text= Number of rows= 0 on slave for a non existent channel for repliation_connection_configuration table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_connection_status where channel_name="non_existent"`
--let $assert_text= Number of rows= 0 on slave for a non existent channel for repliation_connection_status table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_configuration where channel_name="non_existent"`
--let $assert_text= Number of rows= 0 on slave for a non existent channel for repliation_applier_configuration table.
--let $assert_cond= $num= 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_status where channel_name="non_existent"`
--let $assert_text= Number of rows= 0 on slave for a non existent channel for repliation_applier_status table.
--let $assert_cond= $num = 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_status_by_coordinator where channel_name="non_existent"`
--let $assert_text= Number of rows= 0 on slave for a non existent channel for repliation_applier_status_by_coordinaotor table.
--let $assert_cond= $num= 0;
--source include/assert.inc

--let $num= `select count(*) from performance_schema.replication_applier_status_by_worker where channel_name="non_existent"`
--let $assert_text= Number of rows= 0 on slave for a non existent channel for replication_applier_status_by_worker table.
--let $assert_cond= "$num"= 0;
--source include/assert.inc

--echo #
--echo #  Test case 3: Rows on multi sourced slave, shall be 2. In workers table it should be
--echo #               2*@@global.replica_parallel_workers
USE performance_schema;

--let $assert_text= there shall be 2 rows for replication_connection_configuration
--let $assert_cond= [select count(*) from performance_schema.replication_connection_configuration] = 2;
--source include/assert.inc

--let $assert_text= there shall be 2 rows for replication_connection_status
--let $assert_cond= [select count(*) from performance_schema.replication_connection_status] = 2;
--source include/assert.inc

--let $assert_text= there shall be 2 rows for replication_applier_configuration
--let $assert_cond= [select count(*) from performance_schema.replication_applier_configuration] = 2 ;
--source include/assert.inc

--let $assert_text= there shall be 2 rows for replication_applier_status
--let $assert_cond= [select count(*) from performance_schema.replication_applier_status] = 2;
--source include/assert.inc

# Post Bug#20001173 fix in case of Single Threaded Slave(STS) mode SQL
# thread's status will be reported as part of
# performance_schema.replication_applier_status_by_worker table.
--let $w_rows= `select @@global.replica_parallel_workers`
if ($w_rows != 0)
{
  --let $assert_text= there shall be 2 rows for replication_connection_configuration
  --let $assert_cond= [select count(*) from performance_schema.replication_applier_status_by_coordinator] = 2;
  --source include/assert.inc
}

if ($w_rows == 0)
{
  --let $assert_text= there shall be 2 rows for replication_connection_configuration
  --let $assert_cond= [select count(*) from performance_schema.replication_applier_status_by_worker] = 2;
  --source include/assert.inc
}

# Post Bug#20001173 fix in case of Single Threaded Slave(STS) mode SQL
# thread's status will be reported as part of
# performance_schema.replication_applier_status_by_worker table.
# SQL thread will have the corresponding worker_id as 0.
--let $pfs_w_rows= `select count(*) from performance_schema.replication_applier_status_by_worker where worker_id != 0`
--let $w_rows= `select 2*@@global.replica_parallel_workers`
--let $assert_text= there shall be 2*global.replica_parallel_workers rows for 2 configured channels.
--let $assert_cond= "$pfs_w_rows" = "$w_rows"
--source include/assert.inc


--echo #
--echo # Test case 4: Verify that selects work on named channels
--echo #              Do this verification by comparing with some known data
--echo #
#check that port value returned by channel_3 is same as SERVER_MYPORT_3
--let $channel_3_port= `select port from performance_schema.replication_connection_configuration where channel_name="channel_3"`
--let $assert_text= pfs query shall return the correct query value of the channel_3 port.
--let $assert_cond= "$channel_3_port"= "$SERVER_MYPORT_3"
--source include/assert.inc

--let $service_state= `select service_state from performance_schema.replication_connection_status where channel_name="channel_3"`
--let $assert_text= service state of receiver thread shall be ON.
--let $assert_cond= "ON" = "$service_state"
--source include/assert.inc

let $sss_value=query_get_value(SHOW SLAVE STATUS FOR CHANNEL 'channel_3', SQL_Delay, 1);
let $ps_value= query_get_value(select Desired_Delay from performance_schema.replication_applier_configuration where channel_name = "channel_3", Desired_Delay, 1);
let $assert_text= Value returned by SSS and PS table for Desired_Delay should be same.;
let $assert_cond= "$sss_value" = "$ps_value";
--source include/assert.inc

--let $service_state= `select service_state from performance_schema.replication_applier_status where channel_name="channel_3"`
--let $assert_text= service state shall be ON
--let $assert_cond= "ON" = "$service_state"
--source include/assert.inc

# Post Bug#20001173 fix in case of Single Threaded Slave(STS) mode SQL
# thread's status will be reported as part of
# performance_schema.replication_applier_status_by_worker table.
--let $service_state= `select service_state from performance_schema.replication_applier_status_by_worker where channel_name="channel_3"`
--let $assert_text= service state of coordinator shall be ON
--let $assert_cond= "ON" = "$service_state"
--source include/assert.inc

if ( `select @@GLOBAL.replica_parallel_workers > 0`)
{
  --let $service_state= `select service_state from performance_schema.replication_applier_status_by_worker where channel_name="channel_3" and worker_id= 1`
  --let $assert_text= service state of worker thread shall be ON
  --let $assert_cond= "ON" = "$service_state"
  --let $include_silent= 1
  --source include/assert.inc
  --let $include_silent= 0
}


--echo #
--echo #  Stop the slave on channel_1 and verify that rows are still instact
--echo #
--let $rpl_channel_name= 'channel_1'
--source include/stop_slave.inc

--echo #
--echo # Test case 5: Verify that selects work on named channels even
--echo #              after a channel is stopped. test that service state is OFF
--echo #

--let $channel_1_port= `select port from performance_schema.replication_connection_configuration where channel_name="channel_1"`
--let $assert_text=  After stop slave for channel, replication_connection_configuration is intact
--let $assert_cond= "$channel_1_port"= "$SERVER_MYPORT_1"
--source include/assert.inc

--let $service_state= `select service_state from performance_schema.replication_connection_status where channel_name="channel_1"`
--let $assert_text=  After stop slave for channel, service state is OFF for connection_status table.
--let $assert_cond= "OFF" = "$service_state"
--source include/assert.inc

--let $service_state= `select service_state from performance_schema.replication_applier_status where channel_name="channel_1"`
--let $assert_text=  After stop slave for channel, service state is OFF for applier_status table.
--let $assert_cond= "OFF" = "$service_state"
--source include/assert.inc

# Post Bug#20001173 fix in case of Single Threaded Slave(STS) mode SQL
# thread's status will be reported as part of
# performance_schema.replication_applier_status_by_worker table.
--let $service_state= `select service_state from performance_schema.replication_applier_status_by_worker where channel_name="channel_1"`
--let $assert_text=  After stop slave for channel, service state is OFF by replication_applier_status_by_worker table.
--let $assert_cond= "OFF" = "$service_state"
--source include/assert.inc

if ( `select @@GLOBAL.replica_parallel_workers > 0`)
{
  --let $service_state= `select service_state from performance_schema.replication_applier_status_by_worker where channel_name="channel_1" and worker_id= 1`
  --let $assert_text= service state of worker thread shall be OFF
  --let $assert_cond= "OFF" = "$service_state"
  --let $include_silent= 1
  --source include/assert.inc
  --let $include_silent= 0
}


--echo #
--echo #  Test case 6): Do a sanity check that joins can be performed as replication tables
--echo #                 contain multiple rows now.
# Post Bug#20001173 fix in case of Single Threaded Slave(STS) mode SQL
# thread's status will be reported as part of
# performance_schema.replication_applier_status_by_worker table.
--let $w_rows= `select @@global.replica_parallel_workers`
if ($w_rows != 0)
{
  --disable_query_log
  --replace_column 3 # 4 # 6 # 7 # 13 #
  --query_vertical(select connection_status.CHANNEL_NAME, GROUP_NAME, SOURCE_UUID, connection_status.THREAD_ID, connection_status.SERVICE_STATE, COUNT_RECEIVED_HEARTBEATS, LAST_HEARTBEAT_TIMESTAMP, RECEIVED_TRANSACTION_SET, connection_status.LAST_ERROR_NUMBER, connection_status.LAST_ERROR_MESSAGE, connection_status.LAST_ERROR_TIMESTAMP, coordinator_status.CHANNEL_NAME, coordinator_status.THREAD_ID, coordinator_status.SERVICE_STATE, coordinator_status.LAST_ERROR_NUMBER, coordinator_status.LAST_ERROR_MESSAGE, coordinator_status.LAST_ERROR_TIMESTAMP from replication_connection_status connection_status, replication_applier_status_by_coordinator coordinator_status where connection_status.channel_name= coordinator_status.channel_name)
  --enable_query_log
}

if ($w_rows == 0)
{
  --disable_query_log
  --replace_column 3 # 4 # 6 # 7 # 13 #
  --query_vertical(select connection_status.CHANNEL_NAME, GROUP_NAME, SOURCE_UUID, connection_status.THREAD_ID, connection_status.SERVICE_STATE, COUNT_RECEIVED_HEARTBEATS, LAST_HEARTBEAT_TIMESTAMP, RECEIVED_TRANSACTION_SET, connection_status.LAST_ERROR_NUMBER, connection_status.LAST_ERROR_MESSAGE, connection_status.LAST_ERROR_TIMESTAMP, worker_status.CHANNEL_NAME, worker_status.THREAD_ID, worker_status.SERVICE_STATE, worker_status.LAST_ERROR_NUMBER, worker_status.LAST_ERROR_MESSAGE, worker_status.LAST_ERROR_TIMESTAMP from replication_connection_status connection_status, replication_applier_status_by_worker worker_status where connection_status.channel_name= worker_status.channel_name and worker_status.worker_id=0)
  --enable_query_log
}

--echo #
--echo # Test case 7: Delete channel_3 and count the number of rows in each table.
--echo #
--let $rpl_channel_name= 'channel_3'
--source include/stop_slave.inc
RESET SLAVE ALL FOR CHANNEL 'channel_3';

--let $assert_text= there shall be 1 in replication connection_configuration table after a channel is deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_connection_configuration] = 1;
--source include/assert.inc

--let $assert_text= there shall be 1 in replication connection_status table after a channel is deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_connection_status] = 1;
--source include/assert.inc

--let $assert_text= there shall be 1 in replication applier_config table after a channel is deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_applier_configuration] = 1 ;
--source include/assert.inc

--let $assert_text= there shall be 1 in replication applier_status table after a channel is deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_applier_status] = 1;
--source include/assert.inc

# Post Bug#20001173 fix in case of Single Threaded Slave(STS) mode SQL
# thread's status will be reported as part of
# performance_schema.replication_applier_status_by_worker table.
# SQL thread will have the corresponding worker_id as 0.
--let $w_rows= `select @@global.replica_parallel_workers`

if ($w_rows == 0)
{
  --let $assert_text= there shall be 1 applier after a channel is deleted.
  --let $assert_cond= [select count(*) from performance_schema.replication_applier_status_by_worker where worker_id=0] = 1;
  --source include/assert.inc
  --let $pfs_w_rows= `select count(*) from performance_schema.replication_applier_status_by_worker where worker_id !=0`
  --let $w_rows= `select @@global.replica_parallel_workers`
  --let $assert_text= there shall be global.replica_parallel_workers rows after a channel is deleted.
  --let $assert_cond= "$pfs_w_rows" = "$w_rows"
  --source include/assert.inc
}
if ($w_rows != 0)
{
  --let $assert_text= there shall be 1 applier after a channel is deleted.
  --let $assert_cond= [select count(*) from performance_schema.replication_applier_status_by_coordinator] = 1;
  --source include/assert.inc
  --let $pfs_w_rows= `select count(*) from performance_schema.replication_applier_status_by_worker`
  --let $w_rows= `select @@global.replica_parallel_workers`
  --let $assert_text= there shall be global.replica_parallel_workers rows after a channel is deleted.
  --let $assert_cond= "$pfs_w_rows" = "$w_rows"
  --source include/assert.inc
}



--echo #
--echo #  Test case 8: Do a complete delete and see if the number of rows are zero
--echo #
RESET SLAVE ALL;


--let $assert_text= there shall be 0 rows after all channels are deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_connection_configuration] = 0;
--source include/assert.inc

--let $assert_text= there shall be 0 rows after all channels are deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_connection_status] = 0;
--source include/assert.inc

--let $assert_text= there shall be 0 rows after all channels are deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_applier_configuration] = 0 ;
--source include/assert.inc

--let $assert_text= there shall be 0 rows after all channels are deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_applier_status] = 0;
--source include/assert.inc

--let $assert_text= there shall be 0 rows after all channels are deleted.
--let $assert_cond= [select count(*) from performance_schema.replication_applier_status_by_coordinator] = 0;
--source include/assert.inc

--let $pfs_w_rows= `select count(*) from performance_schema.replication_applier_status_by_worker`
--let $w_rows= `select @@global.replica_parallel_workers`
--let $assert_text= there shall be global.replica_parallel_workers rows after a channel is deleted.

--let $assert_text= there shall be 0 rows after all channels are deleted.
--let $assert_cond= "$pfs_w_rows" = "0"
--source include/assert.inc

--let $rpl_skip_sync= 1
--let $rpl_skip_delete_channels= 1
--let $rpl_only_running_threads= 1
--let $rpl_group_replication= 1
--source include/rpl_end.inc