path: root/mysql-test/suite/rpl/t/rpl_parallel.test

--source include/have_innodb.inc
--source include/have_debug.inc
--source include/have_debug_sync.inc
--let $rpl_topology=1->2
--source include/rpl_init.inc

# Test various aspects of parallel replication.

--connection server_2
SET @old_parallel_threads=@@GLOBAL.slave_parallel_threads;
--error ER_SLAVE_MUST_STOP
SET GLOBAL slave_parallel_threads=10;
--source include/stop_slave.inc
SET GLOBAL slave_parallel_threads=10;
CHANGE MASTER TO master_use_gtid=slave_pos;
--source include/start_slave.inc


--echo *** Test long-running query in domain 1 can run in parallel with short queries in domain 0 ***

--connection server_1
ALTER TABLE mysql.gtid_slave_pos ENGINE=InnoDB;
CREATE TABLE t1 (a int PRIMARY KEY) ENGINE=MyISAM;
CREATE TABLE t2 (a int PRIMARY KEY) ENGINE=InnoDB;
INSERT INTO t1 VALUES (1);
INSERT INTO t2 VALUES (1);
--save_master_pos

--connection server_2
--sync_with_master

# Block the table t1 to simulate a replicated query taking a long time.
--connect (con_temp1,127.0.0.1,root,,test,$SERVER_MYPORT_2,)
LOCK TABLE t1 WRITE;

--connection server_1
SET gtid_domain_id=1;
# This query will be blocked on the slave until UNLOCK TABLES.
INSERT INTO t1 VALUES (2);
SET gtid_domain_id=0;
# These t2 queries can be replicated in parallel with the prior t1 query, as
# they are in a separate replication domain.
INSERT INTO t2 VALUES (2);
INSERT INTO t2 VALUES (3);
BEGIN;
INSERT INTO t2 VALUES (4);
INSERT INTO t2 VALUES (5);
COMMIT;
INSERT INTO t2 VALUES (6);

--connection server_2
--let $wait_condition= SELECT COUNT(*) = 6 FROM t2
--source include/wait_condition.inc

SELECT * FROM t2 ORDER by a;

--connection con_temp1
SELECT * FROM t1;
UNLOCK TABLES;

--connection server_2
--let $wait_condition= SELECT COUNT(*) = 2 FROM t1
--source include/wait_condition.inc

SELECT * FROM t1 ORDER BY a;


--echo *** Test two transactions in different domains committed in opposite order on slave but in a single group commit. ***
--connection server_2
--source include/stop_slave.inc

--connection server_1
# Use a stored function to inject a debug_sync into the appropriate THD.
# The function does nothing on the master, and on the slave it injects the
# desired debug_sync action(s).
SET sql_log_bin=0;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;

SET @old_format= @@SESSION.binlog_format;
SET binlog_format='statement';
SET gtid_domain_id=1;
INSERT INTO t2 VALUES (foo(10,
    'commit_before_enqueue SIGNAL ready1 WAIT_FOR cont1',
    'commit_after_release_LOCK_prepare_ordered SIGNAL ready2'));

--connection server_2
FLUSH LOGS;
--source include/wait_for_binlog_checkpoint.inc
SET sql_log_bin=0;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    IF d1 != '' THEN
      SET debug_sync = d1;
    END IF;
    IF d2 != '' THEN
      SET debug_sync = d2;
    END IF;
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;
SET @old_format=@@GLOBAL.binlog_format;
SET GLOBAL binlog_format=statement;
# We need to restart all parallel threads for the new global setting to
# be copied to the session-level values.
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
--source include/start_slave.inc

# First make sure the first insert is ready to commit, but not queued yet.
SET debug_sync='now WAIT_FOR ready1';

--connection server_1
SET gtid_domain_id=2;
INSERT INTO t2 VALUES (foo(11,
    'commit_before_enqueue SIGNAL ready3 WAIT_FOR cont3',
    'commit_after_release_LOCK_prepare_ordered SIGNAL ready4 WAIT_FOR cont4'));
SET gtid_domain_id=0;
SELECT * FROM t2 WHERE a >= 10 ORDER BY a;

--connection server_2
# Now wait for the second insert to queue itself as the leader, and then
# wait for more commits to queue up.
SET debug_sync='now WAIT_FOR ready3';
SET debug_sync='now SIGNAL cont3';
SET debug_sync='now WAIT_FOR ready4';
# Now allow the first insert to queue up to participate in group commit.
SET debug_sync='now SIGNAL cont1';
SET debug_sync='now WAIT_FOR ready2';
# Finally allow the second insert to proceed and do the group commit.
SET debug_sync='now SIGNAL cont4';

--let $wait_condition= SELECT COUNT(*) = 2 FROM t2 WHERE a >= 10
--source include/wait_condition.inc
SELECT * FROM t2 WHERE a >= 10 ORDER BY a;
# The two INSERT transactions should have been committed in opposite order,
# but in the same group commit (seen by precense of cid=# in the SHOW
# BINLOG output).
--let $binlog_file= slave-bin.000002
--source include/show_binlog_events.inc
FLUSH LOGS;
--source include/wait_for_binlog_checkpoint.inc

# Restart all the slave parallel worker threads, to clear all debug_sync actions.
--connection server_2
--source include/stop_slave.inc
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
SET debug_sync='RESET';
--source include/start_slave.inc


--echo *** Test that group-committed transactions on the master can replicate in parallel on the slave. ***
--connection server_1
FLUSH LOGS;
--source include/wait_for_binlog_checkpoint.inc
CREATE TABLE t3 (a INT PRIMARY KEY, b INT) ENGINE=InnoDB;
# Create some sentinel rows so that the rows inserted in parallel fall into
# separate gaps and do not cause gap lock conflicts.
INSERT INTO t3 VALUES (1,1), (3,3), (5,5), (7,7);
--save_master_pos
--connection server_2
--sync_with_master

# We want to test that the transactions can execute out-of-order on
# the slave, but still end up committing in-order, and in a single
# group commit.
#
# The idea is to group-commit three transactions together on the master:
# A, B, and C. On the slave, C will execute the insert first, then A,
# and then B. But B manages to complete before A has time to commit, so
# all three end up committing together.
#
# So we start by setting up some row locks that will block transactions
# A and B from executing, allowing C to run first.

--connection con_temp1
BEGIN;
INSERT INTO t3 VALUES (2,102);
--connect (con_temp2,127.0.0.1,root,,test,$SERVER_MYPORT_2,)
BEGIN;
INSERT INTO t3 VALUES (4,104);

# On the master, queue three INSERT transactions as a single group commit.
--connect (con_temp3,127.0.0.1,root,,test,$SERVER_MYPORT_1,)
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued1 WAIT_FOR master_cont1';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (2, foo(12,
    'commit_after_release_LOCK_prepare_ordered SIGNAL slave_queued1 WAIT_FOR slave_cont1',
    ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued1';

--connect (con_temp4,127.0.0.1,root,,test,$SERVER_MYPORT_1,)
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (4, foo(14,
    'commit_after_release_LOCK_prepare_ordered SIGNAL slave_queued2',
    ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued2';

--connect (con_temp5,127.0.0.1,root,,test,$SERVER_MYPORT_1,)
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued3';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (6, foo(16,
    'group_commit_waiting_for_prior SIGNAL slave_queued3',
    ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued3';
SET debug_sync='now SIGNAL master_cont1';

--connection con_temp3
REAP;
--connection con_temp4
REAP;
--connection con_temp5
REAP;

--connection server_1
SELECT * FROM t3 ORDER BY a;
--let $binlog_file= master-bin.000002
--source include/show_binlog_events.inc

# First, wait until insert 3 is ready to queue up for group commit, but is
# waiting for insert 2 to commit before it can do so itself.
--connection server_2
SET debug_sync='now WAIT_FOR slave_queued3';

# Next, let insert 1 proceed, and allow it to queue up as the group commit
# leader, but let it wait for insert 2 to also queue up before proceeding.
--connection con_temp1
ROLLBACK;
--connection server_2
SET debug_sync='now WAIT_FOR slave_queued1';

# Now let insert 2 proceed and queue up.
--connection con_temp2
ROLLBACK;
--connection server_2
SET debug_sync='now WAIT_FOR slave_queued2';
# And finally, we can let insert 1 proceed and do the group commit with all
# three insert transactions together.
SET debug_sync='now SIGNAL slave_cont1';

# Wait for the commit to complete and check that all three transactions
# group-committed together (will be seen in the binlog as all three having
# cid=# on their GTID event).
--let $wait_condition= SELECT COUNT(*) = 3 FROM t3 WHERE a IN (2,4,6)
--source include/wait_condition.inc
SELECT * FROM t3 ORDER BY a;
--let $binlog_file= slave-bin.000003
--source include/show_binlog_events.inc


--echo *** Test STOP SLAVE in parallel mode ***
--connection server_2
--source include/stop_slave.inc

--connection server_1
# Set up a couple of transactions. The first will be blocked halfway
# through on a lock, and while it is blocked we initiate STOP SLAVE.
# We then test that the halfway-initiated transaction is allowed to
# complete, but no subsequent ones.
# We have to use statement-based mode and set
# binlog_direct_non_transactional_updates=0; otherwise the binlog will
# be split into two event groups, one for the MyISAM part and one for the
# InnoDB part.
SET binlog_direct_non_transactional_updates=0;
SET sql_log_bin=0;
CALL mtr.add_suppression("Statement is unsafe because it accesses a non-transactional table after accessing a transactional table within the same transaction");
SET sql_log_bin=1;
BEGIN;
INSERT INTO t2 VALUES (20);
--disable_warnings
INSERT INTO t1 VALUES (20);
--disable_warnings
INSERT INTO t2 VALUES (21);
INSERT INTO t3 VALUES (20, 20);
COMMIT;
INSERT INTO t3 VALUES(21, 21);
INSERT INTO t3 VALUES(22, 22);
SET binlog_format=@old_format;
--save_master_pos

# Start a connection that will block the replicated transaction halfway.
--connection con_temp1
BEGIN;
INSERT INTO t2 VALUES  (21);

--connection server_2
START SLAVE;
# Wait for the MyISAM change to be visible, after which replication will wait
# for con_temp1 to roll back.
--let $wait_condition= SELECT COUNT(*) = 1 FROM t1 WHERE a=20
--source include/wait_condition.inc

--connection con_temp2
# Initiate slave stop. It will have to wait for the current event group
# to complete.
send STOP SLAVE;

--connection con_temp1
ROLLBACK;

--connection con_temp2
reap;

--connection server_2
--source include/wait_for_slave_to_stop.inc
# We should see the first transaction applied, but not the two others.
SELECT * FROM t1 WHERE a >= 20 ORDER BY a;
SELECT * FROM t2 WHERE a >= 20 ORDER BY a;
SELECT * FROM t3 WHERE a >= 20 ORDER BY a;

--source include/start_slave.inc
--sync_with_master
SELECT * FROM t1 WHERE a >= 20 ORDER BY a;
SELECT * FROM t2 WHERE a >= 20 ORDER BY a;
SELECT * FROM t3 WHERE a >= 20 ORDER BY a;


--connection server_2
# Respawn all worker threads to clear any left-over debug_sync or other stuff.
--source include/stop_slave.inc
SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
--source include/start_slave.inc


--echo *** Test killing slave threads at various wait points ***
--echo *** 1. Test killing transaction waiting in commit for previous transaction to commit ***

# Set up three transactions on the master that will be group-committed
# together so they can be replicated in parallel on the slave.
--connection con_temp3
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued1 WAIT_FOR master_cont1';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (31, foo(31,
    'commit_before_prepare_ordered WAIT_FOR t2_waiting',
    'commit_after_prepare_ordered SIGNAL t1_ready WAIT_FOR t1_cont'));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued1';

--connection con_temp4
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2';
SET binlog_format=statement;
BEGIN;
# This insert is just so we can get T2 to wait while a query is running that we
# can see in SHOW PROCESSLIST so we can get its thread_id to kill later.
INSERT INTO t3 VALUES (32, foo(32,
    'ha_write_row_end SIGNAL t2_query WAIT_FOR t2_cont',
    ''));
# This insert sets up debug_sync points so that T2 will tell when it is at its
# wait point where we want to kill it - and when it has been killed.
INSERT INTO t3 VALUES (33, foo(33,
    'group_commit_waiting_for_prior SIGNAL t2_waiting',
    'group_commit_waiting_for_prior_killed SIGNAL t2_killed'));
send COMMIT;

--connection server_1
SET debug_sync='now WAIT_FOR master_queued2';

--connection con_temp5
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued3';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (34, foo(34,
    '',
    ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued3';
SET debug_sync='now SIGNAL master_cont1';

--connection con_temp3
REAP;
--connection con_temp4
REAP;
--connection con_temp5
REAP;

--connection server_1
SELECT * FROM t3 WHERE a >= 30 ORDER BY a;

--connection server_2
SET sql_log_bin=0;
CALL mtr.add_suppression("Query execution was interrupted");
CALL mtr.add_suppression("Commit failed due to failure of an earlier commit on which this one depends");
CALL mtr.add_suppression("Slave: Connection was killed");
SET sql_log_bin=1;
# Wait until T2 is inside executing its insert of 32, then find it in SHOW
# PROCESSLIST to know its thread id for KILL later.
SET debug_sync='now WAIT_FOR t2_query';
--let $thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%foo(32%' AND INFO NOT LIKE '%LIKE%'`
SET debug_sync='now SIGNAL t2_cont';

# Wait until T2 has entered its wait for T1 to commit, and T1 has
# progressed into its commit phase.
SET debug_sync='now WAIT_FOR t1_ready';

# Now kill the transaction T2.
--replace_result $thd_id THD_ID
eval KILL $thd_id;

# Wait until T2 has reacted on the kill.
SET debug_sync='now WAIT_FOR t2_killed';

# Now we can allow T1 to proceed.
SET debug_sync='now SIGNAL t1_cont';

--let $slave_sql_errno= 1317,1963
--source include/wait_for_slave_sql_error.inc
STOP SLAVE IO_THREAD;
SELECT * FROM t3 WHERE a >= 30 ORDER BY a;

# Now we have to disable the debug_sync statements, so they do not trigger
# when the events are retried.
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;

--connection server_1
INSERT INTO t3 VALUES (39,0);
--save_master_pos

--connection server_2
--source include/start_slave.inc
--sync_with_master
SELECT * FROM t3 WHERE a >= 30 ORDER BY a;
# Restore the foo() function.
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    IF d1 != '' THEN
      SET debug_sync = d1;
    END IF;
    IF d2 != '' THEN
      SET debug_sync = d2;
    END IF;
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;


--connection server_2
# Respawn all worker threads to clear any left-over debug_sync or other stuff.
--source include/stop_slave.inc
SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
--source include/start_slave.inc


--echo *** 2. Same as (1), but without restarting IO thread after kill of SQL threads ***

# Set up three transactions on the master that will be group-committed
# together so they can be replicated in parallel on the slave.
--connection con_temp3
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued1 WAIT_FOR master_cont1';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (41, foo(41,
    'commit_before_prepare_ordered WAIT_FOR t2_waiting',
    'commit_after_prepare_ordered SIGNAL t1_ready WAIT_FOR t1_cont'));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued1';

--connection con_temp4
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2';
SET binlog_format=statement;
BEGIN;
# This insert is just so we can get T2 to wait while a query is running that we
# can see in SHOW PROCESSLIST so we can get its thread_id to kill later.
INSERT INTO t3 VALUES (42, foo(42,
    'ha_write_row_end SIGNAL t2_query WAIT_FOR t2_cont',
    ''));
# This insert sets up debug_sync points so that T2 will tell when it is at its
# wait point where we want to kill it - and when it has been killed.
INSERT INTO t3 VALUES (43, foo(43,
    'group_commit_waiting_for_prior SIGNAL t2_waiting',
    'group_commit_waiting_for_prior_killed SIGNAL t2_killed'));
send COMMIT;

--connection server_1
SET debug_sync='now WAIT_FOR master_queued2';

--connection con_temp5
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued3';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (44, foo(44,
    '',
    ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued3';
SET debug_sync='now SIGNAL master_cont1';

--connection con_temp3
REAP;
--connection con_temp4
REAP;
--connection con_temp5
REAP;

--connection server_1
SELECT * FROM t3 WHERE a >= 40 ORDER BY a;

--connection server_2
# Wait until T2 is inside executing its insert of 42, then find it in SHOW
# PROCESSLIST to know its thread id for KILL later.
SET debug_sync='now WAIT_FOR t2_query';
--let $thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%foo(42%' AND INFO NOT LIKE '%LIKE%'`
SET debug_sync='now SIGNAL t2_cont';

# Wait until T2 has entered its wait for T1 to commit, and T1 has
# progressed into its commit phase.
SET debug_sync='now WAIT_FOR t1_ready';

# Now kill the transaction T2.
--replace_result $thd_id THD_ID
eval KILL $thd_id;

# Wait until T2 has reacted on the kill.
SET debug_sync='now WAIT_FOR t2_killed';

# Now we can allow T1 to proceed.
SET debug_sync='now SIGNAL t1_cont';

--let $slave_sql_errno= 1317,1963
--source include/wait_for_slave_sql_error.inc
SELECT * FROM t3 WHERE a >= 40 ORDER BY a;

# Now we have to disable the debug_sync statements, so they do not trigger
# when the events are retried.
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;

--connection server_1
INSERT INTO t3 VALUES (49,0);
--save_master_pos

--connection server_2
START SLAVE SQL_THREAD;
--sync_with_master
SELECT * FROM t3 WHERE a >= 40 ORDER BY a;
# Restore the foo() function.
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    IF d1 != '' THEN
      SET debug_sync = d1;
    END IF;
    IF d2 != '' THEN
      SET debug_sync = d2;
    END IF;
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;


--connection server_2
# Respawn all worker threads to clear any left-over debug_sync or other stuff.
--source include/stop_slave.inc
SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
--source include/start_slave.inc


--echo *** 3. Same as (2), but not using gtid mode ***

--connection server_2
--source include/stop_slave.inc
CHANGE MASTER TO master_use_gtid=no;
--source include/start_slave.inc

--connection server_1
# Set up three transactions on the master that will be group-committed
# together so they can be replicated in parallel on the slave.
--connection con_temp3
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued1 WAIT_FOR master_cont1';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (51, foo(51,
    'commit_before_prepare_ordered WAIT_FOR t2_waiting',
    'commit_after_prepare_ordered SIGNAL t1_ready WAIT_FOR t1_cont'));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued1';

--connection con_temp4
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2';
SET binlog_format=statement;
BEGIN;
# This insert is just so we can get T2 to wait while a query is running that we
# can see in SHOW PROCESSLIST so we can get its thread_id to kill later.
INSERT INTO t3 VALUES (52, foo(52,
    'ha_write_row_end SIGNAL t2_query WAIT_FOR t2_cont',
    ''));
# This insert sets up debug_sync points so that T2 will tell when it is at its
# wait point where we want to kill it - and when it has been killed.
INSERT INTO t3 VALUES (53, foo(53,
    'group_commit_waiting_for_prior SIGNAL t2_waiting',
    'group_commit_waiting_for_prior_killed SIGNAL t2_killed'));
send COMMIT;

--connection server_1
SET debug_sync='now WAIT_FOR master_queued2';

--connection con_temp5
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued3';
SET binlog_format=statement;
send INSERT INTO t3 VALUES (54, foo(54,
    '',
    ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued3';
SET debug_sync='now SIGNAL master_cont1';

--connection con_temp3
REAP;
--connection con_temp4
REAP;
--connection con_temp5
REAP;

--connection server_1
SELECT * FROM t3 WHERE a >= 50 ORDER BY a;

--connection server_2
# Wait until T2 is inside executing its insert of 52, then find it in SHOW
# PROCESSLIST to know its thread id for KILL later.
SET debug_sync='now WAIT_FOR t2_query';
--let $thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%foo(52%' AND INFO NOT LIKE '%LIKE%'`
SET debug_sync='now SIGNAL t2_cont';

# Wait until T2 has entered its wait for T1 to commit, and T1 has
# progressed into its commit phase.
SET debug_sync='now WAIT_FOR t1_ready';

# Now kill the transaction T2.
--replace_result $thd_id THD_ID
eval KILL $thd_id;

# Wait until T2 has reacted on the kill.
SET debug_sync='now WAIT_FOR t2_killed';

# Now we can allow T1 to proceed.
SET debug_sync='now SIGNAL t1_cont';

--let $slave_sql_errno= 1317,1963
--source include/wait_for_slave_sql_error.inc
SELECT * FROM t3 WHERE a >= 50 ORDER BY a;

# Now we have to disable the debug_sync statements, so they do not trigger
# when the events are retried.
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;

--connection server_1
INSERT INTO t3 VALUES (59,0);
--save_master_pos

--connection server_2
START SLAVE SQL_THREAD;
--sync_with_master
SELECT * FROM t3 WHERE a >= 50 ORDER BY a;
# Restore the foo() function.
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    IF d1 != '' THEN
      SET debug_sync = d1;
    END IF;
    IF d2 != '' THEN
      SET debug_sync = d2;
    END IF;
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;


--source include/stop_slave.inc
CHANGE MASTER TO master_use_gtid=slave_pos;
--source include/start_slave.inc

--connection server_2
# Respawn all worker threads to clear any left-over debug_sync or other stuff.
--source include/stop_slave.inc
SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=3;
--source include/start_slave.inc


--echo *** 4. Test killing thread that is waiting to start transaction until previous transaction commits ***

# We set up four transactions T1, T2, T3, and T4 on the master. T2, T3, and T4
# can run in parallel with each other (same group commit and commit id),
# but not in parallel with T1.
#
# We use three worker threads. T1 and T2 will be queued on the first, T3 on
# the second, and T4 on the third. We will delay T1 commit, T3 will wait for
# T1 to commit before it can start. We will kill T3 during this wait, and
# check that everything works correctly.
#
# It is rather tricky to get the correct thread id of the worker to kill.
# We start by injecting three dummy transactions in a debug_sync-controlled
# manner to be able to get known thread ids for the workers in a pool with
# just 3 worker threads. Then we let in each of the real test transactions
# T1-T4 one at a time in a way which allows us to know which transaction
# ends up with which thread id.

--connection server_1
SET binlog_format=statement;
SET gtid_domain_id=2;
INSERT INTO t3 VALUES (60, foo(60,
       'ha_write_row_end SIGNAL d2_query WAIT_FOR d2_cont2',
       'rpl_parallel_end_of_group SIGNAL d2_done WAIT_FOR d2_cont'));
SET gtid_domain_id=0;

--connection server_2
SET debug_sync='now WAIT_FOR d2_query';
--let $d2_thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%foo(60%' AND INFO NOT LIKE '%LIKE%'`

--connection server_1
SET gtid_domain_id=1;
BEGIN;
# These debug_sync's will linger on and be used to control T3 later.
INSERT INTO t3 VALUES (61, foo(61,
       'rpl_parallel_start_waiting_for_prior SIGNAL t3_waiting',
       'rpl_parallel_start_waiting_for_prior_killed SIGNAL t3_killed'));
INSERT INTO t3 VALUES (62, foo(62,
       'ha_write_row_end SIGNAL d1_query WAIT_FOR d1_cont2',
       'rpl_parallel_end_of_group SIGNAL d1_done WAIT_FOR d1_cont'));
COMMIT;
SET gtid_domain_id=0;

--connection server_2
SET debug_sync='now WAIT_FOR d1_query';
--let $d1_thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%foo(62%' AND INFO NOT LIKE '%LIKE%'`

--connection server_1
SET gtid_domain_id=0;
INSERT INTO t3 VALUES (63, foo(63,
       'ha_write_row_end SIGNAL d0_query WAIT_FOR d0_cont2',
       'rpl_parallel_end_of_group SIGNAL d0_done WAIT_FOR d0_cont'));

--connection server_2
SET debug_sync='now WAIT_FOR d0_query';
--let $d0_thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE INFO LIKE '%foo(63%' AND INFO NOT LIKE '%LIKE%'`

SET debug_sync='now SIGNAL d2_cont2';
SET debug_sync='now WAIT_FOR d2_done';
SET debug_sync='now SIGNAL d1_cont2';
SET debug_sync='now WAIT_FOR d1_done';
SET debug_sync='now SIGNAL d0_cont2';
SET debug_sync='now WAIT_FOR d0_done';

# Now prepare the real transactions T1, T2, T3, T4 on the master.

--connection con_temp3
# Create transaction T1.
SET binlog_format=statement;
INSERT INTO t3 VALUES (64, foo(64,
       'commit_before_prepare_ordered SIGNAL t1_waiting WAIT_FOR t1_cont', ''));

# Create transaction T2, as a group commit leader on the master.
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued2 WAIT_FOR master_cont2';
send INSERT INTO t3 VALUES (65, foo(65, '', ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued2';

--connection con_temp4
# Create transaction T3, participating in T2's group commit.
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued3';
send INSERT INTO t3 VALUES (66, foo(66, '', ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued3';

--connection con_temp5
# Create transaction T4, participating in group commit with T2 and T3.
SET debug_sync='commit_after_release_LOCK_prepare_ordered SIGNAL master_queued4';
send INSERT INTO t3 VALUES (67, foo(67, '', ''));

--connection server_1
SET debug_sync='now WAIT_FOR master_queued4';
SET debug_sync='now SIGNAL master_cont2';

--connection con_temp3
REAP;
--connection con_temp4
REAP;
--connection con_temp5
REAP;

--connection server_1
SELECT * FROM t3 WHERE a >= 60 ORDER BY a;

--connection server_2
# Now we have the four transactions pending for replication on the slave.
# Let them be queued for our three worker threads in a controlled fashion.
# We put them at a stage where T1 is delayed and T3 is waiting for T1 to
# commit before T3 can start. Then we kill T3.

# Make the worker D0 free, and wait for T1 to be queued in it.
SET debug_sync='now SIGNAL d0_cont';
SET debug_sync='now WAIT_FOR t1_waiting';

# T2 will be queued on the same worker D0 as T1.
# Now release worker D1, and wait for T3 to be queued in it.
# T3 will wait for T1 to commit before it can start.
SET debug_sync='now SIGNAL d1_cont';
SET debug_sync='now WAIT_FOR t3_waiting';

# Release worker D2. T4 may or may not have time to be queued on it, but
# it will not be able to complete due to T3 being killed.
SET debug_sync='now SIGNAL d2_cont';

# Now we kill the waiting transaction T3 in worker D1.
--replace_result $d1_thd_id THD_ID
eval KILL $d1_thd_id;

# Wait until T3 has reacted on the kill.
SET debug_sync='now WAIT_FOR t3_killed';

# Now we can allow T1 to proceed.
SET debug_sync='now SIGNAL t1_cont';

--let $slave_sql_errno= 1317,1927,1963
--source include/wait_for_slave_sql_error.inc
STOP SLAVE IO_THREAD;
SELECT * FROM t3 WHERE a >= 60 ORDER BY a;

# Now we have to disable the debug_sync statements, so they do not trigger
# when the events are retried.
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;

--connection server_1
INSERT INTO t3 VALUES (69,0);
--save_master_pos

--connection server_2
--source include/start_slave.inc
--sync_with_master
SELECT * FROM t3 WHERE a >= 60 ORDER BY a;
# Restore the foo() function.
SET sql_log_bin=0;
DROP FUNCTION foo;
--delimiter ||
CREATE FUNCTION foo(x INT, d1 VARCHAR(500), d2 VARCHAR(500))
  RETURNS INT DETERMINISTIC
  BEGIN
    IF d1 != '' THEN
      SET debug_sync = d1;
    END IF;
    IF d2 != '' THEN
      SET debug_sync = d2;
    END IF;
    RETURN x;
  END
||
--delimiter ;
SET sql_log_bin=1;

--connection server_2
# Respawn all worker threads to clear any left-over debug_sync or other stuff.
--source include/stop_slave.inc
SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
--source include/start_slave.inc


--echo *** 5. Test killing thread that is waiting for queue of max length to shorten ***

--let $wait_condition= SELECT COUNT(*) = 1 FROM INFORMATION_SCHEMA.PROCESSLIST WHERE STATE LIKE '%Slave has read all relay log%'
--source include/wait_condition.inc
--let $thd_id= `SELECT ID FROM INFORMATION_SCHEMA.PROCESSLIST WHERE STATE LIKE '%Slave has read all relay log%'`
SET @old_max_queued= @@GLOBAL.slave_parallel_max_queued;
SET GLOBAL slave_parallel_max_queued=9000;

--connection server_1
--let bigstring= `SELECT REPEAT('x', 10000)`
SET binlog_format=statement;
# Create an event that will wait to be signalled.
INSERT INTO t3 VALUES (70, foo(0,
       'ha_write_row_end SIGNAL query_waiting WAIT_FOR query_cont', ''));
--disable_query_log
# Create an event that will fill up the queue.
eval INSERT INTO t3 VALUES (71, LENGTH('$bigstring'));
--enable_query_log

--connection server_2
SET debug_sync='now WAIT_FOR query_waiting';
# Inject that the SQL driver thread will signal `wait_queue_ready' to debug_sync
# as it goes to wait for the event queue to become smaller than the value of
# @@slave_parallel_max_queued.
SET @old_dbug= @@GLOBAL.debug_dbug;
SET GLOBAL debug_dbug="+d,rpl_parallel_wait_queue_max";

--connection server_1
# This event will have to wait for the queue to become shorter before it can
# be queued. We will test that things work when we kill the SQL driver thread
# during this wait.
INSERT INTO t3 VALUES (72, 0);
SELECT * FROM t3 WHERE a >= 70 ORDER BY a;

--connection server_2
SET debug_sync='now WAIT_FOR wait_queue_ready';

--replace_result $thd_id THD_ID
eval KILL $thd_id;

SET debug_sync='now WAIT_FOR wait_queue_killed';
SET debug_sync='now SIGNAL query_cont';

--let $slave_sql_errno= 1317,1927,1963
--source include/wait_for_slave_sql_error.inc
STOP SLAVE IO_THREAD;
SELECT * FROM t3 WHERE a >= 70 ORDER BY a;

SET GLOBAL debug_dbug=@old_dbug;
SET GLOBAL slave_parallel_max_queued= @old_max_queued;

--connection server_1
INSERT INTO t3 VALUES (73,0);
--save_master_pos

--connection server_2
--source include/start_slave.inc
--sync_with_master
SELECT * FROM t3 WHERE a >= 70 ORDER BY a;


--connection server_2
--source include/stop_slave.inc
SET GLOBAL binlog_format=@old_format;
SET GLOBAL slave_parallel_threads=0;
SET GLOBAL slave_parallel_threads=10;
--source include/start_slave.inc


--connection server_2
--source include/stop_slave.inc
SET GLOBAL slave_parallel_threads=@old_parallel_threads;
--source include/start_slave.inc
SET DEBUG_SYNC= 'RESET';

--connection server_1
DROP function foo;
DROP TABLE t1,t2,t3;
SET DEBUG_SYNC= 'RESET';

--source include/rpl_end.inc