3286 Andrei Elkin 2011-06-12
wl#5569 MTS
making default.push to run rpl suite with non-default --mts-slave-parallel-workers > 0 in all three format/mode (row,stmt, mixed). The default is run for all suites in mixed mode and rpl suites with row+ps, stmt formats.
modified:
mysql-test/collections/default.push
3285 Andrei Elkin 2011-06-12 [merge]
wl#5569 MTS
manual merge with few fixes for segfault of the last merge from the trunk etc, compilation issue on embedded.
added:
mysql-test/suite/rpl/r/rpl_parallel_seconds_behind_master.result
mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master-slave.opt
mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master.test
modified:
mysql-test/extra/rpl_tests/rpl_parallel_benchmark_load.test
mysql-test/suite/rpl/r/rpl_parallel_benchmark.result
mysql-test/suite/rpl/r/rpl_slave_grp_exec.result
mysql-test/suite/rpl/t/rpl_bug26395.test
mysql-test/suite/rpl/t/rpl_cross_version.test
mysql-test/suite/rpl/t/rpl_packet.test
mysql-test/suite/rpl/t/rpl_seconds_behind_master.test
mysql-test/suite/rpl/t/rpl_slave_grp_exec.test
sql/binlog.cc
sql/events.cc
sql/log_event.cc
sql/log_event.h
sql/rpl_rli.cc
sql/rpl_rli.h
sql/rpl_rli_pdb.cc
sql/rpl_rli_pdb.h
sql/rpl_slave.cc
sql/share/errmsg-utf8.txt
sql/sp.cc
sql/sql_acl.cc
sql/sql_class.h
sql/sql_db.cc
sql/sql_rename.cc
sql/sql_table.cc
sql/sql_trigger.cc
sql/sql_view.cc
storage/blackhole/ha_blackhole.cc
3284 Alfranio Correia 2011-06-09
Post-fixes for merge.
Fixed compilation in Windows and removed an used options.
modified:
mysql-test/extra/rpl_tests/rpl_crash_safe.inc
mysql-test/suite/rpl/r/rpl_row_crash_safe.result
mysql-test/suite/rpl/r/rpl_stm_mixed_crash_safe.result
mysql-test/suite/rpl/t/rpl_mixed_parallel_crash_safe-slave.opt
mysql-test/suite/rpl/t/rpl_row_crash_safe-slave.opt
mysql-test/suite/rpl/t/rpl_row_parallel_crash_safe-slave.opt
mysql-test/suite/rpl/t/rpl_stm_mixed_crash_safe-slave.opt
mysql-test/suite/rpl/t/rpl_stm_parallel_crash_safe-slave.opt
sql/mysqld.cc
sql/mysqld.h
sql/rpl_info_factory.cc
sql/sql_base.cc
=== modified file 'mysql-test/collections/default.push'
--- a/mysql-test/collections/default.push 2011-03-23 11:37:04 +0000
+++ b/mysql-test/collections/default.push 2011-06-12 19:33:32 +0000
@@ -1,3 +1,11 @@
-perl mysql-test-run.pl --timer --force --parallel=auto --comment=n_mix --vardir=var-n_mix --mysqld=--binlog-format=mixed --experimental=collections/default.experimental --skip-ndb --skip-test-list=collections/disabled-per-push.list --unit-tests
-perl mysql-test-run.pl --timer --force --parallel=auto --comment=main_ps_row --vardir=var-main-ps_row --suite=main --ps-protocol --mysqld=--binlog-format=row --experimental=collections/default.experimental --skip-ndb --skip-test-list=collections/disabled-per-push.list
-perl mysql-test-run.pl --timer --force --parallel=auto --comment=main_embedded --vardir=var-main_emebbed --suite=main --embedded --experimental=collections/default.experimental --skip-ndb
+perl mysql-test-run.pl --timer --force --parallel=auto --comment=n_mix --vardir=var-n_mix --mysqld=--binlog-format=mixed --experimental=collections/default.experimental --skip-ndb --unit-tests
+
+perl mysql-test-run.pl --timer --force --parallel=auto --comment=rpl_ps_row --vardir=var-rpl-ps_row --ps-protocol --mysqld=--binlog-format=row --experimental=collections/default.experimental --skip-ndb --suite=rpl
+
+perl mysql-test-run.pl --timer --force --parallel=auto --comment=rpl_stmt --vardir=var-rpl-stmt --mysqld=--binlog-format=statement --experimental=collections/default.experimental --skip-ndb --suite=rpl
+
+perl mysql-test-run.pl --timer --force --parallel=auto --comment=rpl_n_mix_MTS --vardir=var-rpl-n_mix --mysqld=--binlog-format=mixed --experimental=collections/default.experimental --skip-ndb --unit-tests --mysqld=--mts-slave-parallel-workers=4 --mysqld=--slave-transaction-retries=0 --suite=rpl
+
+perl mysql-test-run.pl --timer --force --parallel=auto --comment=rpl_ps_row_MTS --vardir=var-rpl-ps_row --ps-protocol --mysqld=--binlog-format=row --experimental=collections/default.experimental --skip-ndb --mysqld=--mts-slave-parallel-workers=4 --mysqld=--slave-transaction-retries=0 --suite=rpl
+
+perl mysql-test-run.pl --timer --force --parallel=auto --comment=rpl_stmt_rpl_MTS --vardir=var-rpl-stmt --mysqld=--binlog-format=statement --experimental=collections/default.experimental --skip-ndb --mysqld=--mts-slave-parallel-workers=4 --mysqld=--slave-transaction-retries=0 --suite=rpl
=== modified file 'mysql-test/extra/rpl_tests/rpl_parallel_benchmark_load.test'
--- a/mysql-test/extra/rpl_tests/rpl_parallel_benchmark_load.test 2011-06-05 17:01:51 +0000
+++ b/mysql-test/extra/rpl_tests/rpl_parallel_benchmark_load.test 2011-06-12 17:36:17 +0000
@@ -6,13 +6,13 @@
# load volume parameter
#
-let $iter= 08;
-let $tables= 4;
-let $wk_i_queries= 4;
+let $iter= 02;
+let $tables= 2;
+let $wk_i_queries= 2;
let $wk_m_queries= 0;
let $nk_i_queries= 0;
let $nk_m_queries= 0;
-let $pre_inserted_rows= 200;
+let $pre_inserted_rows= 50;
connection slave;
@@ -264,14 +264,22 @@ let $wait_timeout= 600;
let $wait_condition= SELECT count(*)+sleep(1) = 5 FROM test1.benchmark;
source include/wait_condition.inc;
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+
+# cleanup for files that could not be removed in the end of previous invocation.
+--remove_files_wildcard $MYSQLD_DATADIR *.out
+
use test;
-select * from test1.benchmark into outfile 'benchmark.out';
+--replace_result $MYSQLD_DATADIR MYSQLD_DATADIR
+eval select * from test1.benchmark into outfile '$MYSQLD_DATADIR/benchmark.out';
select ts from test1.benchmark where state like 'master started load' into @m_0;
select ts from test1.benchmark where state like 'master ends load' into @m_1;
select ts from test1.benchmark where state like 'slave takes on load' into @s_0;
select ts from test1.benchmark where state like 'slave ends load' into @s_1;
-select time_to_sec(@m_1) - time_to_sec(@m_0) as 'delta_m',
- time_to_sec(@s_1) - time_to_sec(@s_0) as 'delta_s' into outfile 'delta.out';
+--replace_result $MYSQLD_DATADIR MYSQLD_DATADIR
+eval select time_to_sec(@m_1) - time_to_sec(@m_0) as 'delta_m',
+ time_to_sec(@s_1) - time_to_sec(@s_0) as 'delta_s' into outfile '$MYSQLD_DATADIR/delta.out';
--enable_result_log
--enable_query_log
=== modified file 'mysql-test/suite/rpl/r/rpl_parallel_benchmark.result'
--- a/mysql-test/suite/rpl/r/rpl_parallel_benchmark.result 2011-04-02 11:32:02 +0000
+++ b/mysql-test/suite/rpl/r/rpl_parallel_benchmark.result 2011-06-12 17:36:17 +0000
@@ -6,12 +6,12 @@ include/stop_slave.inc
start slave;
stop slave sql_thread;
use test;
-select * from test1.benchmark into outfile 'benchmark.out';
+select * from test1.benchmark into outfile 'MYSQLD_DATADIR/benchmark.out';
select ts from test1.benchmark where state like 'master started load' into @m_0;
select ts from test1.benchmark where state like 'master ends load' into @m_1;
select ts from test1.benchmark where state like 'slave takes on load' into @s_0;
select ts from test1.benchmark where state like 'slave ends load' into @s_1;
select time_to_sec(@m_1) - time_to_sec(@m_0) as 'delta_m',
-time_to_sec(@s_1) - time_to_sec(@s_0) as 'delta_s' into outfile 'delta.out';
+time_to_sec(@s_1) - time_to_sec(@s_0) as 'delta_s' into outfile 'MYSQLD_DATADIR/delta.out';
set @@global.mts_exp_slave_local_timestamp= @save.mts_exp_slave_local_timestamp;
include/rpl_end.inc
=== added file 'mysql-test/suite/rpl/r/rpl_parallel_seconds_behind_master.result'
--- a/mysql-test/suite/rpl/r/rpl_parallel_seconds_behind_master.result 1970-01-01 00:00:00 +0000
+++ b/mysql-test/suite/rpl/r/rpl_parallel_seconds_behind_master.result 2011-06-10 08:04:00 +0000
@@ -0,0 +1,35 @@
+include/master-slave.inc
+[connection master]
+call mtr.add_suppression('Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT');
+include/stop_slave.inc
+set @save.mts_slave_parallel_workers= @@global.mts_slave_parallel_workers;
+set @@global.mts_slave_parallel_workers= 4;
+set @save.mts_checkpoint_period= @@global.mts_checkpoint_period;
+set @@global.mts_checkpoint_period= 500;
+set @save.slave_transaction_retries= @@global.slave_transaction_retries;
+start slave io_thread;
+include/wait_for_slave_io_to_start.inc
+create table t1 (f1 int) engine=innodb;
+create table t2 (f1 int) engine=innodb;
+begin;
+insert into t1 values (sleep(3) + 1);
+Warnings:
+Note 1592 Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT. Statement is unsafe because it uses a system function that may return a different value on the slave.
+commit;
+begin;
+insert into t2 values (1);
+commit;
+include/start_slave.inc
+include/sync_slave_io_with_master.inc
+lock table t2 write;
+First transaction is done, now get Seconds_Behind_Master after it...
+include/wait_for_slave_param.inc [Seconds_Behind_Master]
+Seconds_Behind_Master after first transaction is as expected.
+unlock tables;
+include/wait_for_slave_param.inc [Seconds_Behind_Master]
+drop tables t1, t2;
+set @@global.mts_checkpoint_period= @save.mts_checkpoint_period;
+set @@global.mts_slave_parallel_workers= @save.mts_slave_parallel_workers;
+set @@global.slave_transaction_retries= @save.slave_transaction_retries;
+include/rpl_end.inc
+end of the tests
=== modified file 'mysql-test/suite/rpl/r/rpl_slave_grp_exec.result'
--- a/mysql-test/suite/rpl/r/rpl_slave_grp_exec.result 2011-02-23 11:54:58 +0000
+++ b/mysql-test/suite/rpl/r/rpl_slave_grp_exec.result 2011-06-12 19:05:05 +0000
@@ -30,7 +30,7 @@ SELECT * FROM t3 ORDER BY a;
a b
1 ZZ
call mtr.add_suppression("Slave SQL.*Table .test.t3. doesn.t exist.* Error_code: 1146");
-include/wait_for_slave_sql_error.inc [errno=1146]
+include/wait_for_slave_sql_to_stop.inc
SHOW TABLES LIKE 't%';
Tables_in_test (t%)
t1
@@ -58,7 +58,7 @@ INSERT INTO t3 VALUES(2, 'B');
INSERT INTO t2 VALUES(2, 'B');
INSERT INTO t1 VALUES(2, 'B');
UPDATE t1 SET b = 'X' WHERE a = 2;
-include/wait_for_slave_sql_error.inc [errno=1146]
+include/wait_for_slave_sql_to_stop.inc
SELECT * FROM t1 ORDER BY a;
a b
2 X
@@ -94,7 +94,7 @@ INSERT INTO t1 VALUES (3, 'C'), (4, 'D')
INSERT INTO t2 VALUES (3, 'C'), (4, 'D');
INSERT INTO t3 VALUES (3, 'C'), (4, 'D');
COMMIT;
-include/wait_for_slave_sql_error.inc [errno=1146]
+include/wait_for_slave_sql_to_stop.inc
SELECT * FROM t1 ORDER BY a;
a b
3 C
=== modified file 'mysql-test/suite/rpl/t/rpl_bug26395.test'
--- a/mysql-test/suite/rpl/t/rpl_bug26395.test 2010-12-19 17:07:28 +0000
+++ b/mysql-test/suite/rpl/t/rpl_bug26395.test 2011-06-08 20:18:08 +0000
@@ -37,6 +37,10 @@ source include/have_innodb.inc;
source include/have_debug.inc;
source include/master-slave.inc;
+# test adapts simulation of incomplete transaction that MTS does not tolerate
+# when is stopped. So it reacts with an error whereas the single-threaded is fine.
+-- source include/not_mts_slave_parallel_workers.inc
+
--echo ==== Initialize ====
@@ -67,7 +71,6 @@ source include/sync_slave_io_with_master
# Sync slave's SQL thread.
sync_with_master 0;
-
--echo ==== Verify results on slave ====
source include/stop_slave.inc;
=== modified file 'mysql-test/suite/rpl/t/rpl_cross_version.test'
--- a/mysql-test/suite/rpl/t/rpl_cross_version.test 2011-01-17 18:51:01 +0000
+++ b/mysql-test/suite/rpl/t/rpl_cross_version.test 2011-06-12 19:05:05 +0000
@@ -12,6 +12,14 @@
--source include/master-slave.inc
+# The test is disabled for windows due to
+# Bug #42879 CHANGE MASTER RELAY_LOG_FILE=path fails on windows
+# Todo: release it from not_windows
+--source include/not_windows.inc
+
+# EXEC_LOAD_EVENT of 4.1 binlog can't be supported
+-- source include/not_mts_slave_parallel_workers.inc
+
#
# Bug#31240 load data infile replication between (4.0 or 4.1) and 5.1 fails
#
=== modified file 'mysql-test/suite/rpl/t/rpl_packet.test'
--- a/mysql-test/suite/rpl/t/rpl_packet.test 2011-06-09 15:27:47 +0000
+++ b/mysql-test/suite/rpl/t/rpl_packet.test 2011-06-12 19:05:05 +0000
@@ -11,6 +11,13 @@
# max-out size db name
source include/master-slave.inc;
source include/have_binlog_format_row.inc;
+
+# TODO: Fixing is handed over to Sergei.
+# The test runs slow in MTS mode because of state of MTS at time of the graceful stop.
+# In this case MTS can't stop immediately if there is a Worker that received a BEGIN but never COMMIT.
+-- source include/not_mts_slave_parallel_workers.inc
+
+
call mtr.add_suppression("Slave I/O: Got a packet bigger than 'max_allowed_packet' bytes, Error_code: 1153");
call mtr.add_suppression("Slave I/O: Got fatal error 1236 from master when reading data from binary log:");
@@ -283,6 +290,7 @@ eval SET @@global.max_allowed_packet= $o
DROP TABLE t1;
# Clear Last_IO_Error
+
--source include/stop_slave_sql.inc
RESET SLAVE;
=== added file 'mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master-slave.opt'
--- a/mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master-slave.opt 1970-01-01 00:00:00 +0000
+++ b/mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master-slave.opt 2011-06-12 17:36:17 +0000
@@ -0,0 +1 @@
+--slave-transaction-retries=0
=== added file 'mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master.test'
--- a/mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master.test 1970-01-01 00:00:00 +0000
+++ b/mysql-test/suite/rpl/t/rpl_parallel_seconds_behind_master.test 2011-06-12 17:36:17 +0000
@@ -0,0 +1,141 @@
+#
+# Testing of Seconds_Behind_Master (SBM) to behave as the following
+# SBM changes discretely per some number of processed group of events,
+# e.g transactions.
+# The number is either @@global.slave_checkpoint_group or less if
+# @@global.slave_checkpoint_period timer elapses first.
+# The value updates *after* the last group commit is executed.
+# Resetting to zero behavior when Slave goes to read events is
+# preserved.
+#
+
+-- source include/master-slave.inc
+-- source include/have_debug.inc
+
+# make the test only run once (STMT is actually needed because we rely
+# on SHOW PROCESS LIST output in some of the tests)
+-- source include/have_binlog_format_statement.inc
+
+# todo: sleep() is deterministic!!!
+call mtr.add_suppression('Unsafe statement written to the binary log using statement format since BINLOG_FORMAT = STATEMENT');
+
+# restart slave in MTS mode
+connection slave;
+source include/stop_slave.inc;
+set @save.mts_slave_parallel_workers= @@global.mts_slave_parallel_workers;
+set @@global.mts_slave_parallel_workers= 4;
+set @save.mts_checkpoint_period= @@global.mts_checkpoint_period;
+set @@global.mts_checkpoint_period= 500; # 0.5 sec
+# to avoid warnings
+set @save.slave_transaction_retries= @@global.slave_transaction_retries;
+
+#
+# Idea of demonstration.
+#
+# Two transactions on the same db are generated for MTS. The first has some
+# natural delay, and the 2nd will be locked out of its data.
+# While the first is being processed Coordinator empties the relay-log
+# to enter waiting for more events.
+# It does so being awakened @@global.slave_checkpoint_period-ically
+# to check out if some groups of events have been done to update SBM (demo 1).
+#
+# Resetting of SBM is a possibility whenever Coordinator
+# notices no more groups left neither to read nor to process (demo 2).
+#
+
+start slave io_thread;
+source include/wait_for_slave_io_to_start.inc;
+
+connection master;
+
+create table t1 (f1 int) engine=innodb;
+create table t2 (f1 int) engine=innodb;
+save_master_pos;
+
+# 1st transaction that will be used to compute SBM.
+# It will hang $idle time at least in processing to get min estimate for SBM
+let $idle= 3;
+
+begin;
+ eval insert into t1 values (sleep($idle) + 1);
+commit;
+
+# 2nd transaction to block Coordinator from resetting SBM
+begin;
+ insert into t2 values (1);
+commit;
+
+connection slave;
+source include/start_slave.inc;
+
+#
+# all events are in relay-log
+#
+connection master;
+
+source include/sync_slave_io_with_master.inc;
+
+# tables are created on slave
+connection slave;
+sync_with_master;
+
+connection slave1;
+lock table t2 write; # to block 2nd trans
+
+connection slave;
+
+let $count= 1;
+let $table= t1;
+source include/wait_until_rows_count.inc;
+
+--echo First transaction is done, now get Seconds_Behind_Master after it...
+
+#
+# demo 1:
+# So now 1st is over, and it SBM must have gain at least $idle seconds.
+#
+let $slave_param= Seconds_Behind_Master;
+let $slave_param_comparison= >=;
+let $slave_param_value= $idle;
+source include/wait_for_slave_param.inc;
+
+# checking sanity of SBM value
+let $sbm= query_get_value("SHOW SLAVE STATUS", Seconds_Behind_Master, 1);
+let $lower_bound= $idle;
+if (`SELECT $sbm < $idle`)
+{
+ -- echo *** Seconds behind master is $sbm less than minimum expected $idle ***
+ -- source include/show_rpl_debug_info.inc
+ -- die
+}
+
+--echo Seconds_Behind_Master after first transaction is as expected.
+
+connection slave1;
+unlock tables;
+
+#
+# demo 2: Resetting of SBM upon all groups have been processed and
+# nothing left in relay-log.
+#
+connection slave;
+let $slave_param= Seconds_Behind_Master;
+let $slave_param_comparison= =;
+let $slave_param_value= 0;
+source include/wait_for_slave_param.inc;
+
+##
+# cleanup
+##
+connection master;
+drop tables t1, t2;
+
+sync_slave_with_master;
+
+set @@global.mts_checkpoint_period= @save.mts_checkpoint_period;
+set @@global.mts_slave_parallel_workers= @save.mts_slave_parallel_workers;
+set @@global.slave_transaction_retries= @save.slave_transaction_retries;
+
+--source include/rpl_end.inc
+
+--echo end of the tests
=== modified file 'mysql-test/suite/rpl/t/rpl_seconds_behind_master.test'
--- a/mysql-test/suite/rpl/t/rpl_seconds_behind_master.test 2010-12-19 17:22:30 +0000
+++ b/mysql-test/suite/rpl/t/rpl_seconds_behind_master.test 2011-06-10 08:04:00 +0000
@@ -5,6 +5,10 @@
# on SHOW PROCESS LIST output in some of the tests)
-- source include/have_binlog_format_statement.inc
+# this test logics requires adaption to MTS policies for SBM
+# (see rpl_parallel_seconds_behind_master.test)
+-- source include/not_mts_slave_parallel_workers.inc
+
SET @old_debug= @@global.debug;
call mtr.add_suppression('Unsafe statement binlogged in statement format since BINLOG_FORMAT = STATEMENT');
=== modified file 'mysql-test/suite/rpl/t/rpl_slave_grp_exec.test'
--- a/mysql-test/suite/rpl/t/rpl_slave_grp_exec.test 2011-02-23 11:54:58 +0000
+++ b/mysql-test/suite/rpl/t/rpl_slave_grp_exec.test 2011-06-12 19:05:05 +0000
@@ -23,6 +23,12 @@
--source include/master-slave.inc
--echo
+# Test is MTS unfriendly because of
+# a. incompatible with STS error reporting (MTS stop due to an error in applying
+# causes inconsistency so the latter is reported)
+# b. failing recovery
+-- source include/not_mts_slave_parallel_workers.inc
+
# Create tables and data
--echo *** Preparing data ***
--connection master
@@ -62,10 +68,22 @@ SELECT * FROM t2 ORDER BY a;
SELECT * FROM t3 ORDER BY a;
--connection slave
-# 1146 = ER_NO_SUCH_TABLE
call mtr.add_suppression("Slave SQL.*Table .test.t3. doesn.t exist.* Error_code: 1146");
---let $slave_sql_errno= 1146
---source include/wait_for_slave_sql_error.inc
+
+# in MTS case error is either of two:
+#--let $slave_sql_errno= 1146,1593
+# whereas in the single-threaded case:
+# 1146 = ER_NO_SUCH_TABLE
+#--let $slave_sql_errno= 1146
+
+--source include/wait_for_slave_sql_to_stop.inc
+let $slave_sql_errno= query_get_value(SHOW SLAVE STATUS, Last_SQL_Errno, 1);
+if (`select $slave_sql_errno != 1146 and $slave_sql_errno != 1593`)
+{
+ --echo Unexpected error: $slave_sql_errno
+ --die
+}
+
SHOW TABLES LIKE 't%';
if (`SELECT @@BINLOG_FORMAT = 'ROW'`) {
--replace_regex /AA/AA_for_row_or_XX_for_stmt_mixed/
@@ -111,8 +129,18 @@ UPDATE t1 SET b = 'X' WHERE a = 2;
--connection slave
# 1146 = ER_NO_SUCH_TABLE
---let $slave_sql_errno= 1146
---source include/wait_for_slave_sql_error.inc
+# in MTS case error is either of two:
+#--let $slave_sql_errno= 1146,1593
+# whereas in the single-threaded case:
+#--let $slave_sql_errno= 1146
+
+--source include/wait_for_slave_sql_to_stop.inc
+let $slave_sql_errno= query_get_value(SHOW SLAVE STATUS, Last_SQL_Errno, 1);
+if (`select $slave_sql_errno != 1146 and $slave_sql_errno != 1593`)
+{
+ --echo Unexpected error: $slave_sql_errno
+ --die
+}
--connection master
SELECT * FROM t1 ORDER BY a;
@@ -126,6 +154,8 @@ SELECT * FROM t2 ORDER BY a;
--source include/stop_slave_io.inc
RENAME TABLE t3_bak TO t3;
+
+# TODO: recovery. Alfranio it fails to recover here.
--source include/start_slave.inc
--connection master
@@ -157,8 +187,19 @@ COMMIT;
--connection slave
# 1146 = ER_NO_SUCH_TABLE
---let $slave_sql_errno= 1146
---source include/wait_for_slave_sql_error.inc
+# in MTS case error is either of two:
+#--let $slave_sql_errno= 1146,1593
+# whereas in the single-threaded case:
+#--let $slave_sql_errno= 1146
+
+--source include/wait_for_slave_sql_to_stop.inc
+let $slave_sql_errno= query_get_value(SHOW SLAVE STATUS, Last_SQL_Errno, 1);
+if (`select $slave_sql_errno != 1146 and $slave_sql_errno != 1593`)
+{
+ --echo Unexpected error: $slave_sql_errno
+ --die
+}
+
--connection master
SELECT * FROM t1 ORDER BY a;
=== modified file 'sql/binlog.cc'
--- a/sql/binlog.cc 2011-06-09 15:27:47 +0000
+++ b/sql/binlog.cc 2011-06-12 19:05:05 +0000
@@ -4945,26 +4945,37 @@ THD::binlog_set_pending_rows_event(Rows_
/**
@param db db name c-string to be inserted into abc-sorted
THD::binlog_accessed_db_names list.
-
- Note, as the list node data (explicitly) so the node
- struct itself (implicitly) are allocated in
- thd->mem_root to be cleared at the end of the query
- processing (@c THD::cleanup_after_query()).
+
+ Note, as the list node data so the node
+ struct itself are allocated in THD::main_mem_root.
+ The list lasts for the top-level query time and resets
+ in @c THD::cleanup_after_query() and Query_log_event::write().
*/
void
-THD::add_to_binlog_updated_dbs(const char *db)
+THD::add_to_binlog_accessed_dbs(const char *db)
{
char *after_db;
+ MEM_ROOT *db_mem_root= &main_mem_root;
+
+ if (!binlog_accessed_db_names)
+ binlog_accessed_db_names= new (db_mem_root) List<char>;
+
if (binlog_accessed_db_names->elements > MAX_DBS_IN_EVENT_MTS)
{
push_warning_printf(this, MYSQL_ERROR::WARN_LEVEL_WARN,
- ER_UPDATED_DBS_GREATER_MAX,
- ER(ER_UPDATED_DBS_GREATER_MAX),
+ ER_MTS_UPDATED_DBS_GREATER_MAX,
+ ER(ER_MTS_UPDATED_DBS_GREATER_MAX),
MAX_DBS_IN_EVENT_MTS);
return;
}
- after_db= strdup_root(mem_root, db);
+ after_db= strdup_root(db_mem_root, db);
+
+ /*
+ sorted insertion is implemented with first rearranging data
+ (pointer to char*) of the links and final appending of the least
+ ordered data to create a new link in the list.
+ */
if (binlog_accessed_db_names->elements != 0)
{
List_iterator<char> it(*get_binlog_accessed_db_names());
@@ -4991,7 +5002,7 @@ THD::add_to_binlog_updated_dbs(const cha
}
}
if (after_db)
- binlog_accessed_db_names->push_back(after_db);
+ binlog_accessed_db_names->push_back(after_db, &main_mem_root);
}
@@ -5218,27 +5229,6 @@ int THD::decide_logging_format(TABLE_LIS
prev_access_table= table->table;
}
-
- /*
- Master side of DML in the STMT format events parallelization.
- All involving table db:s are stored in a abc-ordered name list.
- In case the number of databases exceeds MAX_DBS_IN_EVENT_MTS maximum
- the list gathering breaks since it won't be sent to the slave.
- */
- if (is_write && variables.binlog_format != BINLOG_FORMAT_ROW &&
- lex->sql_command != SQLCOM_END /* rows-event applying by slave */)
- {
- if (!binlog_accessed_db_names)
- {
- binlog_accessed_db_names= new List<char>; /* thd->mem_root is used */
- }
- for (TABLE_LIST *table= tables; table; table= table->next_global)
- {
- if (table->placeholder())
- continue;
- add_to_binlog_updated_dbs(table->db);
- }
- }
DBUG_PRINT("info", ("flags_write_all_set: 0x%llx", flags_write_all_set));
DBUG_PRINT("info", ("flags_write_some_set: 0x%llx", flags_write_some_set));
@@ -5373,6 +5363,23 @@ int THD::decide_logging_format(TABLE_LIS
DBUG_PRINT("info", ("decision: no logging since an error was generated"));
DBUG_RETURN(-1);
}
+
+ if (is_write && !is_current_stmt_binlog_format_row() &&
+ lex->sql_command != SQLCOM_END /* rows-event applying by slave */)
+ {
+ /*
+ Master side of DML in the STMT format events parallelization.
+ All involving table db:s are stored in a abc-ordered name list.
+ In case the number of databases exceeds MAX_DBS_IN_EVENT_MTS maximum
+ the list gathering breaks since it won't be sent to the slave.
+ */
+ for (TABLE_LIST *table= tables; table; table= table->next_global)
+ {
+ if (table->placeholder())
+ continue;
+ add_to_binlog_accessed_dbs(table->db);
+ }
+ }
DBUG_PRINT("info", ("decision: logging in %s format",
is_current_stmt_binlog_format_row() ?
"ROW" : "STATEMENT"));
=== modified file 'sql/events.cc'
--- a/sql/events.cc 2011-06-09 15:27:47 +0000
+++ b/sql/events.cc 2011-06-12 19:05:05 +0000
@@ -389,7 +389,7 @@ Events::create_event(THD *thd, Event_par
}
else
{
- thd->add_one_db_to_binlog_updated_dbs(parse_data->dbname.str);
+ thd->add_to_binlog_accessed_dbs(parse_data->dbname.str);
/*
If the definer is not set or set to CURRENT_USER, the value of CURRENT_USER
will be written into the binary log as the definer for the SQL thread.
@@ -511,10 +511,9 @@ Events::update_event(THD *thd, Event_par
/* Binlog the alter event. */
DBUG_ASSERT(thd->query() && thd->query_length());
- thd->set_binlog_accessed_db_names(new List<char>);
- thd->add_to_binlog_updated_dbs(parse_data->dbname.str);
+ thd->add_to_binlog_accessed_dbs(parse_data->dbname.str);
if (new_dbname)
- thd->add_to_binlog_updated_dbs(new_dbname->str);
+ thd->add_to_binlog_accessed_dbs(new_dbname->str);
ret= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
}
@@ -583,7 +582,7 @@ Events::drop_event(THD *thd, LEX_STRING
/* Binlog the drop event. */
DBUG_ASSERT(thd->query() && thd->query_length());
- thd->add_one_db_to_binlog_updated_dbs(dbname.str);
+ thd->add_to_binlog_accessed_dbs(dbname.str);
ret= write_bin_log(thd, TRUE, thd->query(), thd->query_length());
}
/* Restore the state of binlog format */
=== modified file 'sql/log_event.cc'
--- a/sql/log_event.cc 2011-06-09 15:27:47 +0000
+++ b/sql/log_event.cc 2011-06-12 19:05:05 +0000
@@ -672,6 +672,9 @@ Log_event::Log_event(enum_event_cache_ty
:temp_buf(0), exec_time(0), flags(0), event_cache_type(cache_type_arg),
event_logging_type(logging_type_arg), crc(0), thd(0),
checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF)
+#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
+ , m_mts_event_ends_group(FALSE)
+#endif
{
server_id= ::server_id;
/*
@@ -694,6 +697,9 @@ Log_event::Log_event(const char* buf,
event_cache_type(EVENT_INVALID_CACHE),
event_logging_type(EVENT_INVALID_LOGGING),
crc(0), checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF)
+#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
+ , m_mts_event_ends_group(FALSE)
+#endif
{
#ifndef MYSQL_CLIENT
thd = 0;
@@ -2356,7 +2362,7 @@ Log_event::continue_group(Relay_log_info
bool Log_event::contains_partition_info()
{
- return get_type_code() == TABLE_MAP_EVENT ||
+ return (get_type_code() == TABLE_MAP_EVENT) ||
(get_type_code() == QUERY_EVENT && !ends_group() && !starts_group()) ||
(get_type_code() == EXECUTE_LOAD_QUERY_EVENT);
}
@@ -2405,6 +2411,7 @@ Slave_worker *Log_event::get_slave_worke
bool is_b_event;
int num_dbs= 0;
Slave_worker *ret_worker= NULL;
+ THD *thd= rli->info_thd;
/* checking partioning properties and perform corresponding actions */
@@ -2466,7 +2473,7 @@ Slave_worker *Log_event::get_slave_worke
{
int i= 0;
num_dbs= mts_number_dbs();
- List_iterator<char> it(*mts_get_dbs(rli->info_thd->mem_root));
+ List_iterator<char> it(*mts_get_dbs(thd->mem_root));
it++;
ret_worker= rli->last_assigned_worker;
@@ -2488,12 +2495,10 @@ Slave_worker *Log_event::get_slave_worke
get_type_code() == QUERY_EVENT,
ret_worker)))
{
- // destroy buffered events of the current group prior to exit
- for (uint k= 0; k < rli->curr_group_da.elements; k++)
- {
- delete *(Log_event**) dynamic_array_ptr(&rli->curr_group_da, k);
- }
-
+ rli->report(ERROR_LEVEL, ER_MTS_CANT_PARALLEL,
+ ER(ER_MTS_CANT_PARALLEL),
+ get_type_str(), rli->get_event_relay_log_name(),
+ rli->get_event_relay_log_pos());
return ret_worker;
}
@@ -2516,13 +2521,13 @@ Slave_worker *Log_event::get_slave_worke
DBUG_ASSERT(i == num_dbs || num_dbs == OVER_MAX_DBS_IN_EVENT_MTS);
- // TODO: convert to C's private mem_root.
+ // TODO: convert to C's private mem_root to reset not per event but rather realrely.
// Releasing the Coord's mem-root from the updated dbs. It's safe to do at this
// point because the root is no longer needed along remained part of Coordinator's
// execution flow.
- free_root(rli->info_thd->mem_root, MYF(MY_KEEP_PREALLOC));
+ free_root(thd->mem_root, MYF(MY_KEEP_PREALLOC));
}
else
{
@@ -2534,15 +2539,25 @@ Slave_worker *Log_event::get_slave_worke
DBUG_ASSERT(rli->curr_group_assigned_parts.elements > 0 ||
ret_worker->id == 0);
}
- else // int_, rand_, user_ var:s
+ else // int_, rand_, user_ var:s, load-data events
{
Log_event *ptr_curr_ev= this;
- DBUG_ASSERT(get_type_code() == INTVAR_EVENT ||
- get_type_code() == RAND_EVENT ||
- get_type_code() == USER_VAR_EVENT ||
- get_type_code() == ROWS_QUERY_LOG_EVENT ||
- get_type_code() == BEGIN_LOAD_QUERY_EVENT);
+ if (!(get_type_code() == INTVAR_EVENT ||
+ get_type_code() == RAND_EVENT ||
+ get_type_code() == USER_VAR_EVENT ||
+ get_type_code() == ROWS_QUERY_LOG_EVENT ||
+ get_type_code() == BEGIN_LOAD_QUERY_EVENT ||
+ get_type_code() == APPEND_BLOCK_EVENT))
+ {
+ DBUG_ASSERT(!ret_worker);
+
+ rli->report(ERROR_LEVEL, ER_MTS_CANT_PARALLEL,
+ ER(ER_MTS_CANT_PARALLEL),
+ get_type_str(), rli->get_event_relay_log_name(),
+ rli->get_event_relay_log_pos());
+ return ret_worker;
+ }
insert_dynamic(&rli->curr_group_da, (uchar*) &ptr_curr_ev);
@@ -2626,9 +2641,12 @@ Slave_worker *Log_event::get_slave_worke
strcpy(ptr_g->checkpoint_relay_log_name,
rli->get_group_relay_log_name());
ptr_g->checkpoint_relay_log_pos= rli->get_group_relay_log_pos();
+ ptr_g->shifted= ret_worker->bitmap_shifted;
+ ret_worker->bitmap_shifted= 0;
ret_worker->checkpoint_notified= TRUE;
}
ptr_g->checkpoint_seqno= rli->checkpoint_seqno;
+ ptr_g->ts= when + (time_t) exec_time;
rli->checkpoint_seqno++;
}
@@ -2815,14 +2833,17 @@ void append_item_to_jobs(slave_job_item
can't be decided yet. In the single threaded sequential mode the
event maps to SQL thread rli.
+ @note in case of MTS failure Coordinator destroys all gathered
+ deferred events.
+
@return 0 as success, otherwise a failure.
*/
int Log_event::apply_event(Relay_log_info const *rli)
{
DBUG_ENTER("LOG_EVENT:apply_event");
- Slave_worker *w= NULL;
Relay_log_info *c_rli= const_cast<Relay_log_info*>(rli); // constless alias
bool parallel= FALSE, async_event= FALSE, seq_event= FALSE;
+ THD *thd= c_rli->info_thd;
worker= c_rli;
@@ -2875,19 +2896,15 @@ int Log_event::apply_event(Relay_log_inf
wrappped with BEGIN/COMMIT or preceeded by User|Int|Random- var.
MTS has to stop to suggest restart in the permanent sequential mode.
*/
+ rli->report(ERROR_LEVEL, ER_MTS_CANT_PARALLEL,
+ ER(ER_MTS_CANT_PARALLEL),
+ get_type_str(), c_rli->get_event_relay_log_name(),
+ c_rli->get_event_relay_log_pos());
+
+ /* Coordinator cant continue, it marks MTS group status accordingly */
+ c_rli->mts_group_status= Relay_log_info::MTS_KILLED_GROUP;
- // TODO: improve err msg
- rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
- ER(ER_SLAVE_FATAL_ERROR),
- "Can't execute all binlog event in parallel mode");
-
- // destroy possible buffered events of the current group prior to exit
- for (uint k= 0; k < rli->curr_group_da.elements; k++)
- {
- delete *(Log_event**) dynamic_array_ptr(&rli->curr_group_da, k);
- }
-
- DBUG_RETURN(-1);
+ goto err;
}
/*
Marking sure the event will be executed in sequential mode.
@@ -2920,13 +2937,28 @@ int Log_event::apply_event(Relay_log_inf
worker= NULL;
c_rli->mts_group_status= Relay_log_info::MTS_IN_GROUP;
- c_rli->last_assigned_worker= w= get_slave_worker_id(c_rli);
- if (!w || DBUG_EVALUATE_IF("fault_injection_get_slave_worker", 1, 0))
- DBUG_RETURN(rli->curr_group_assigned_parts.elements == 0 ? FALSE : TRUE);
+ worker= (Relay_log_info*)
+ (c_rli->last_assigned_worker= get_slave_worker_id(c_rli));
+
+err:
+ if (thd->is_error())
+ {
+ DBUG_ASSERT(!worker);
- worker= (Relay_log_info*) w;
+ // destroy buffered events of the current group prior to exit
+ for (uint k= 0; k < rli->curr_group_da.elements; k++)
+ {
+ delete *(Log_event**) dynamic_array_ptr(&rli->curr_group_da, k);
+ }
+ }
+ else
+ {
+ DBUG_ASSERT(worker || rli->curr_group_assigned_parts.elements == 0);
+ }
- DBUG_RETURN(FALSE);
+ DBUG_RETURN((!thd->is_error() ||
+ DBUG_EVALUATE_IF("fault_injection_get_slave_worker", 1, 0)) ?
+ 0 : -1);
}
@@ -3845,7 +3877,7 @@ Query_log_event::Query_log_event(const c
auto_increment_increment(1), auto_increment_offset(1),
time_zone_len(0), lc_time_names_number(0), charset_database_number(0),
table_map_for_update(0), master_data_written(0),
- mts_accessed_dbs(OVER_MAX_DBS_IN_EVENT_MTS), m_mts_query_ends_group(FALSE)
+ mts_accessed_dbs(OVER_MAX_DBS_IN_EVENT_MTS)
{
ulong data_len;
uint32 tmp;
@@ -6596,19 +6628,19 @@ int Rotate_log_event::do_update_pos(Rela
if ((server_id != ::server_id || rli->replicate_same_server_id) &&
!is_relay_log_event() &&
((!rli->is_parallel_exec() && !rli->is_in_group()) ||
- rli->mts_group_status == Relay_log_info::MTS_NOT_IN_GROUP))
+ rli->mts_group_status != Relay_log_info::MTS_IN_GROUP))
{
+ if (rli->is_parallel_exec())
+ {
+ (void) mts_checkpoint_routine(rli, 0, FALSE, TRUE); // todo: error branch
+ }
+
mysql_mutex_lock(&rli->data_lock);
DBUG_PRINT("info", ("old group_master_log_name: '%s' "
"old group_master_log_pos: %lu",
rli->get_group_master_log_name(),
(ulong) rli->get_group_master_log_pos()));
- if (rli->is_parallel_exec())
- {
- (void) mts_checkpoint_routine(rli, 0, FALSE, TRUE); // todo: error branch
- }
-
memcpy((void *)rli->get_group_master_log_name(),
new_log_ident, ident_len + 1);
rli->notify_group_master_log_name_update();
@@ -6621,7 +6653,7 @@ int Rotate_log_event::do_update_pos(Rela
mysql_mutex_unlock(&rli->data_lock);
rli->flush_info(TRUE); // todo: error branch
if (rli->is_parallel_exec())
- rli->reset_notified_checkpoint();
+ rli->reset_notified_checkpoint(0, when + (time_t) exec_time);
/*
Reset thd->variables.option_bits and sql_mode etc, because this could be the signal of
=== modified file 'sql/log_event.h'
--- a/sql/log_event.h 2011-06-09 15:27:47 +0000
+++ b/sql/log_event.h 2011-06-12 19:05:05 +0000
@@ -1138,37 +1138,6 @@ public:
{
return thd ? thd->db : 0;
}
-
- /*
- The method returns a list of updated by the event databases.
- Other than in the case of Query-log-event the list is just one item.
- */
- virtual List<char>* mts_get_dbs(MEM_ROOT *mem_root)
- {
- List<char> *res= new List<char>;
- res->push_back(strdup_root(mem_root, get_db()));
- return res;
- }
-
- /*
- returns the number of updated by the event databases.
- In other than Query-log-event case that's one.
- */
- virtual uint8 mts_number_dbs() { return 1; }
-
- /*
- Event can be exceptionally marked to force its execution.
- in isolation from any other Workers.
- Other than Query-log-event class should not have any implementation
- of this method.
- */
- virtual void mts_do_isolate_event() { DBUG_ASSERT(0); }
-
- /*
- Verifying whether event is marked to execute in isolation.
- */
- virtual bool mts_is_event_isolated() { return FALSE; }
-
#else
Log_event(enum_event_cache_type cache_type_arg= EVENT_INVALID_CACHE,
enum_event_logging_type logging_type_arg= EVENT_INVALID_LOGGING)
@@ -1300,6 +1269,10 @@ public:
@note There are incompatile combinations such the referred event
is wrapped with BEGIN/COMMIT. Such cases should be identified
by the caller and treates as an error.
+
+ Notice, even though the func returns TRUE, some events
+ like old LOAD-DATA rooted EXEC_LOAD_EVENT can't run even
+ in isolated parallel mode and MTS would have to stop.
@return TRUE if despite permanent parallel execution mode an event
needs applying in a real isolation that is sequentially.
@@ -1313,11 +1286,9 @@ public:
get_type_code() == LOAD_EVENT ||
get_type_code() == SLAVE_EVENT ||
get_type_code() == CREATE_FILE_EVENT ||
- get_type_code() == APPEND_BLOCK_EVENT ||
- get_type_code() == EXEC_LOAD_EVENT ||
get_type_code() == DELETE_FILE_EVENT ||
get_type_code() == NEW_LOAD_EVENT ||
-
+ get_type_code() == EXEC_LOAD_EVENT ||
get_type_code() == FORMAT_DESCRIPTION_EVENT||
get_type_code() == INCIDENT_EVENT;
@@ -1361,6 +1332,45 @@ public:
*/
Slave_worker *get_slave_worker_id(Relay_log_info *rli);
+ /*
+ The method returns a list of updated by the event databases.
+ Other than in the case of Query-log-event the list is just one item.
+ */
+ virtual List<char>* mts_get_dbs(MEM_ROOT *mem_root)
+ {
+ List<char> *res= new List<char>;
+ res->push_back(strdup_root(mem_root, get_db()));
+ return res;
+ }
+
+ /*
+ returns the number of updated by the event databases.
+ In other than Query-log-event case that's one.
+ */
+ virtual uint8 mts_number_dbs() { return 1; }
+
+ /*
+ Event can be exceptionally marked to force its execution.
+ in isolation from any other Workers.
+ Other than Query-log-event class should not have any implementation
+ of this method.
+ */
+ /*
+ Event can be indentified as a group terminator and such fact
+ is memoried by the function.
+ */
+ virtual void mts_do_isolate_event()
+ {
+ DBUG_ASSERT(get_type_code() == QUERY_EVENT ||
+ get_type_code() == EXEC_LOAD_EVENT ||
+ get_type_code() == EXECUTE_LOAD_QUERY_EVENT);
+ m_mts_event_ends_group= TRUE;
+ }
+ /*
+ Verifying whether event is marked to execute in isolation.
+ */
+ virtual bool mts_is_event_isolated() { return m_mts_event_ends_group; }
+
/**
Apply the event to the database.
@@ -1494,6 +1504,8 @@ protected:
non-zero. The caller shall decrease the counter by one.
*/
virtual enum_skip_reason do_shall_skip(Relay_log_info *rli);
+
+ bool m_mts_event_ends_group;
#endif
};
@@ -1951,13 +1963,6 @@ public:
uchar mts_accessed_dbs;
char mts_accessed_db_names[MAX_DBS_IN_EVENT_MTS][NAME_LEN];
- /*
- Event can be indentified as a group terminator and such fact
- is memoried by the function.
- */
- virtual void mts_do_isolate_event() { m_mts_query_ends_group= TRUE; }
- virtual bool mts_is_event_isolated() { return m_mts_query_ends_group; }
-
#ifdef MYSQL_SERVER
Query_log_event(THD* thd_arg, const char* query_arg, ulong query_length,
@@ -2068,9 +2073,6 @@ public: /* !!! Public in this pat
(!strncasecmp(query, STRING_WITH_LEN("ROLLBACK"))
&& strncasecmp(query, STRING_WITH_LEN("ROLLBACK TO ")));
}
-private:
-
- bool m_mts_query_ends_group;
};
@@ -3152,8 +3154,17 @@ public:
bool write(IO_CACHE* file);
const char* get_db() { return db; }
#endif
+#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
/* MTS executes this event sequentially */
virtual uint8 mts_number_dbs() { return OVER_MAX_DBS_IN_EVENT_MTS; }
+ virtual List<char>* mts_get_dbs(MEM_ROOT *mem_root)
+ {
+ List<char> *res= new List<char>;
+ res->push_back(strdup_root(mem_root, ""));
+ return res;
+ }
+#endif
+
private:
#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
virtual int do_apply_event(Relay_log_info const *rli);
=== modified file 'sql/rpl_rli.cc'
--- a/sql/rpl_rli.cc 2011-06-09 15:27:47 +0000
+++ b/sql/rpl_rli.cc 2011-06-12 19:05:05 +0000
@@ -197,17 +197,35 @@ void Relay_log_info::reset_notified_rela
Worker notices the new checkpoint value at the group commit
to reset the current bitmap and starts using the clean bitmap
indexed from zero of being reset checkpoint_seqno.
+ New seconds_behind_master timestamp is installed.
+
+ @param shift number of bits to shift by Worker due to the current
+ checkpoint change.
+ @param new_ts new seconds_behind_master timestamp value unless zero.
+ Zero could be due to FD event.
*/
-void Relay_log_info::reset_notified_checkpoint()
+void Relay_log_info::reset_notified_checkpoint(ulong shift, time_t new_ts= 0)
{
if (!is_parallel_exec())
return;
for (uint i= 0; i < workers.elements; i++)
{
Slave_worker *w= *(Slave_worker **) dynamic_array_ptr(&workers, i);
- w->checkpoint_notified= FALSE;
+ if (w->checkpoint_notified)
+ {
+ w->bitmap_shifted= 0;
+ w->checkpoint_notified= FALSE;
+ }
+ w->bitmap_shifted += shift; // to reset at passing the accumulate value into GAQ
}
checkpoint_seqno= 0;
+
+ if (new_ts)
+ {
+ mysql_mutex_lock(&data_lock);
+ last_master_timestamp= new_ts;
+ mysql_mutex_unlock(&data_lock);
+ }
}
static inline int add_relay_log(Relay_log_info* rli,LOG_INFO* linfo)
@@ -1044,7 +1062,7 @@ void Relay_log_info::stmt_done(my_off_t
flush_info(is_transactional() ? TRUE : FALSE); // Alfranio todo: error branch
if (is_parallel_exec())
- reset_notified_checkpoint();
+ reset_notified_checkpoint(0);
}
}
=== modified file 'sql/rpl_rli.h'
--- a/sql/rpl_rli.h 2011-06-09 15:27:47 +0000
+++ b/sql/rpl_rli.h 2011-06-12 19:05:05 +0000
@@ -496,7 +496,8 @@ public:
includes Single-Threaded-Slave case.
*/
MTS_NOT_IN_GROUP,
- MTS_IN_GROUP, /* at least one event was scheduled to a Worker */
+
+ MTS_IN_GROUP, /* at least one not-terminal event scheduled to a Worker */
MTS_END_GROUP, /* the last scheduled event is a terminal event */
MTS_KILLED_GROUP /* Coordinator gave out to reach MTS_END_GROUP */
} mts_group_status;
@@ -540,7 +541,7 @@ public:
Coordinator notifies Workers about this event. Coordinator and Workers
maintain a bitmap of executed group that is reset with a new checkpoint.
*/
- void reset_notified_checkpoint();
+ void reset_notified_checkpoint(ulong, time_t);
/**
Helper function to do after statement completion.
=== modified file 'sql/rpl_rli_pdb.cc'
--- a/sql/rpl_rli_pdb.cc 2011-06-06 10:51:19 +0000
+++ b/sql/rpl_rli_pdb.cc 2011-06-12 17:36:17 +0000
@@ -225,6 +225,7 @@ bool Slave_worker::commit_positions(Log_
my_free(ptr_g->checkpoint_relay_log_name);
ptr_g->checkpoint_relay_log_name= NULL;
+ // TODO: shift `group_execed' << ptr_g->shifted
bitmap_clear_all(&group_execed);
}
// extract an updated relay-log name to store in Worker's rli.
@@ -725,6 +726,8 @@ Slave_worker *get_least_occupied_worker(
void Slave_worker::slave_worker_ends_group(Log_event* ev, int error)
{
+ DBUG_ENTER("Slave_worker::slave_worker_ends_group");
+
if (!error)
{
ulong gaq_idx= ev->mts_group_cnt;
@@ -797,6 +800,9 @@ void Slave_worker::slave_worker_ends_gro
#ifndef DBUG_OFF
// TODO: open it! DBUG_ASSERT(usage_partition || !entry->worker->jobs.len);
#endif
+ DBUG_PRINT("info",
+ ("Notifying entry %p release by worker %lu", entry, this->id));
+
mysql_cond_signal(&slave_worker_hash_cond);
}
}
@@ -827,6 +833,8 @@ void Slave_worker::slave_worker_ends_gro
mysql_mutex_unlock(&c_rli->info_thd->LOCK_thd_data);
mysql_mutex_unlock(&slave_worker_hash_lock);
}
+
+ DBUG_VOID_RETURN;
}
@@ -1105,6 +1113,9 @@ int wait_for_workers_to_finish(Relay_log
THD *thd= rli->info_thd;
const char info_format[]=
"Waiting for Slave Worker %d to release partition `%s`";
+
+ DBUG_ENTER("wait_for_workers_to_finish");
+
for (uint i= 0, ret= 0; i < hash->records; i++)
{
db_worker_hash_entry *entry;
@@ -1127,16 +1138,22 @@ int wait_for_workers_to_finish(Relay_log
if (entry->usage > 0 && !thd->killed)
{
+ long w_id= entry->worker->id;
sprintf(wait_info, info_format, entry->worker->id, entry->db);
entry->worker= NULL; // mark Worker to signal when usage drops to 0
-
- proc_info= thd->enter_cond(&slave_worker_hash_cond, &slave_worker_hash_lock,
- wait_info);
- mysql_cond_wait(&slave_worker_hash_cond, &slave_worker_hash_lock);
+ do
+ {
+ proc_info= thd->enter_cond(&slave_worker_hash_cond,
+ &slave_worker_hash_lock,
+ wait_info);
+ mysql_cond_wait(&slave_worker_hash_cond, &slave_worker_hash_lock);
+ DBUG_PRINT("info",
+ ("Either got awakened of notified: "
+ "entry %p, usage %lu, worker %lu",
+ entry, entry->usage, w_id));
+ } while (entry->usage != 0 && !thd->killed);
thd->exit_cond(proc_info);
ret++;
-
- DBUG_ASSERT(entry->usage == 0 || thd->killed);
}
else
{
@@ -1150,5 +1167,5 @@ int wait_for_workers_to_finish(Relay_log
if (!ignore)
const_cast<Relay_log_info*>(rli)->mts_group_status= Relay_log_info::MTS_NOT_IN_GROUP;
- return ret;
+ DBUG_RETURN(ret);
}
=== modified file 'sql/rpl_rli_pdb.h'
--- a/sql/rpl_rli_pdb.h 2011-06-05 17:01:51 +0000
+++ b/sql/rpl_rli_pdb.h 2011-06-10 08:04:00 +0000
@@ -144,6 +144,8 @@ typedef struct st_slave_job_group
my_off_t checkpoint_relay_log_pos; // T-event lop_pos filled by W for CheckPoint
char* checkpoint_relay_log_name;
volatile uchar done; // Flag raised by W, read and reset by C
+ ulong shifted; // shift the last CP bitmap at receiving a new CP
+ time_t ts; // Group's timestampt to update Seconds_behind_master
} Slave_job_group;
#define retrieve_job(from, to) \
@@ -265,6 +267,7 @@ public:
long usage_partition; // number of different partitions handled by this worker
volatile bool relay_log_change_notified; // Coord sets and resets, W can read
volatile bool checkpoint_notified; // Coord sets and resets, W can read
+ ulong bitmap_shifted; // shift the last bitmap at receiving new CP
bool wq_overrun_set; // W monitors its queue usage to incr/decr rli->mts_wqs_overrun
/*
We need to make this a dynamic field. /Alfranio
=== modified file 'sql/rpl_slave.cc'
--- a/sql/rpl_slave.cc 2011-06-09 15:27:47 +0000
+++ b/sql/rpl_slave.cc 2011-06-12 19:05:05 +0000
@@ -1053,6 +1053,12 @@ static bool io_slave_killed(THD* thd, Ma
In the event of deffering decision @rli->last_event_start_time waiting
timer is set to force the killed status be accepted upon its expiration.
+ Notice Multi-Threaded-Slave behaives similarly in that when it's being
+ stopped and the current group of assigned events has not yet scheduled
+ completely, Coordinator deferres to accept to leave its read-distribute
+ state. The above timeout ensures waiting won't last endlessly, and in
+ such case an error is repoted.
+
@param thd pointer to a THD instance
@param rli pointer to Relay_log_info instance
@@ -1062,15 +1068,19 @@ static bool io_slave_killed(THD* thd, Ma
static bool sql_slave_killed(THD* thd, Relay_log_info* rli)
{
bool ret= FALSE;
+ bool is_parallel_group= FALSE;
+
DBUG_ENTER("sql_slave_killed");
DBUG_ASSERT(rli->info_thd == thd);
DBUG_ASSERT(rli->slave_running == 1);
if (abort_loop || thd->killed || rli->abort_slave)
{
- if ((!rli->is_parallel_exec() &&
- thd->transaction.all.cannot_safely_rollback() && rli->is_in_group())
- || (rli->mts_group_status == Relay_log_info::MTS_IN_GROUP))
+ if ((is_parallel_group= (rli->is_parallel_exec() &&
+ rli->mts_group_status == Relay_log_info::MTS_IN_GROUP))
+ ||
+ (!rli->is_parallel_exec() &&
+ thd->transaction.all.cannot_safely_rollback() && rli->is_in_group()))
{
char msg_stopped[]=
"... Slave SQL Thread stopped with incomplete event group "
@@ -1116,23 +1126,27 @@ static bool sql_slave_killed(THD* thd, R
if (ret == 0)
{
rli->report(WARNING_LEVEL, 0,
+ !is_parallel_group ?
"Request to stop slave SQL Thread received while "
"applying a group that has non-transactional "
- "changes; waiting for completion of the group ... ");
+ "changes; waiting for completion of the group ... "
+ :
+ "Coordinator thread of multi-threaded slave is being stopped "
+ "in the middle of assigning a group of events; "
+ "deferring to exit until the group completion ... ");
}
else
{
rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
ER(ER_SLAVE_FATAL_ERROR),
- rli->mts_group_status == Relay_log_info::MTS_NOT_IN_GROUP ?
- msg_stopped : msg_stopped_mts);
+ !is_parallel_group ? msg_stopped : msg_stopped_mts);
}
}
else
{
ret= TRUE;
rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR, ER(ER_SLAVE_FATAL_ERROR),
- msg_stopped);
+ !is_parallel_group ? msg_stopped : msg_stopped_mts);
}
}
else
@@ -2977,7 +2991,8 @@ static int exec_relay_log_event(THD* thd
event) or ev->when is set to 0, or a FD from master, we don't update the
last_master_timestamp.
*/
- if (!(ev->is_artificial_event() || ev->is_relay_log_event() ||
+ if (!(rli->is_parallel_exec() ||
+ ev->is_artificial_event() || ev->is_relay_log_event() ||
(ev->when == 0) || ev->get_type_code() == FORMAT_DESCRIPTION_EVENT))
{
rli->last_master_timestamp= ev->when + (time_t) ev->exec_time;
@@ -3719,6 +3734,12 @@ pthread_handler_t handle_slave_worker(vo
{
error= slave_worker_exec_job(w, rli);
}
+
+ /*
+ Cleanup after an error requires clear_error() go first.
+ Otherwise assert(!all) in binlog_rollback()
+ */
+ thd->clear_error();
w->cleanup_context(thd, error);
mysql_mutex_lock(&w->jobs_lock);
@@ -3758,7 +3779,6 @@ err:
if (thd)
{
- thd->clear_error();
mysql_mutex_lock(&LOCK_thread_count);
THD_CHECK_SENTRY(thd);
/*
@@ -4040,7 +4060,7 @@ bool mts_checkpoint_routine(Relay_log_in
} // end of commit_positions
*/
- rli->reset_notified_checkpoint();
+ rli->reset_notified_checkpoint(cnt, rli->gaq->lwm.ts);
end:
set_timespec_nsec(rli->last_clock, 0);
@@ -4083,6 +4103,7 @@ int slave_start_single_worker(Relay_log_
w->curr_group_exec_parts.elements= 0;
w->relay_log_change_notified= FALSE; // the 1st group to contain relaylog name
w->checkpoint_notified= FALSE;
+ w-> bitmap_shifted= 0;
w->workers= rli->workers; // shallow copying is sufficient
w->this_worker= w;
w->wait_jobs= w->trans_jobs= w->stmt_jobs= w->curr_jobs= 0;
@@ -5845,6 +5866,9 @@ static Log_event* next_event(Relay_log_i
rli->set_future_event_relay_log_pos(my_b_tell(cur_log));
ev->future_event_relay_log_pos= rli->get_future_event_relay_log_pos();
+ if (hot_log)
+ mysql_mutex_unlock(log_lock);
+
/*
MTS checkpoint in the successful read branch
*/
@@ -5852,11 +5876,10 @@ static Log_event* next_event(Relay_log_i
if (rli->is_parallel_exec() && (mts_checkpoint_period != 0 || force))
{
ulonglong period= static_cast<ulonglong>(mts_checkpoint_period * 1000000ULL);
+ mysql_mutex_unlock(&rli->data_lock);
mts_checkpoint_routine(rli, period, force, TRUE); // TODO: ALFRANIO ERROR
+ mysql_mutex_lock(&rli->data_lock);
}
-
- if (hot_log)
- mysql_mutex_unlock(log_lock);
DBUG_RETURN(ev);
}
DBUG_ASSERT(thd==rli->info_thd);
@@ -5902,7 +5925,8 @@ static Log_event* next_event(Relay_log_i
/* shows zero while it is sleeping (and until the next event
is about to be executed) */
- rli->last_master_timestamp= 0;
+ if (!rli->is_parallel_exec())
+ rli->last_master_timestamp= 0;
DBUG_ASSERT(rli->relay_log.get_open_count() ==
rli->cur_log_old_open_count);
@@ -5983,7 +6007,12 @@ static Log_event* next_event(Relay_log_i
do
{
+ mysql_mutex_unlock(log_lock);
mts_checkpoint_routine(rli, period, FALSE, FALSE); // TODO: ALFRANIO ERROR
+ mysql_mutex_lock(log_lock);
+ if (rli->gaq->empty())
+ rli->last_master_timestamp= 0;
+
set_timespec_nsec(waittime, period);
thd->enter_cond(log_cond, log_lock,
"Slave has read all relay log; "
@@ -6485,8 +6514,8 @@ int start_slave(THD* thd , Master_info*
{
mi->rli->opt_slave_parallel_workers= 0;
push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
- ER_NO_FEATURE_ON_PARALLEL_SLAVE,
- ER(ER_NO_FEATURE_ON_PARALLEL_SLAVE),
+ ER_MTS_FEATURE_IS_NOT_SUPPORTED,
+ ER(ER_MTS_FEATURE_IS_NOT_SUPPORTED),
"UNTIL condtion",
"Slave is started in the sequential execution mode.");
}
@@ -6498,8 +6527,8 @@ int start_slave(THD* thd , Master_info*
if (mi->rli->opt_slave_parallel_workers != 0 && slave_trans_retries != 0)
{
push_warning_printf(thd, MYSQL_ERROR::WARN_LEVEL_NOTE,
- ER_NO_FEATURE_ON_PARALLEL_SLAVE,
- ER(ER_NO_FEATURE_ON_PARALLEL_SLAVE),
+ ER_MTS_FEATURE_IS_NOT_SUPPORTED,
+ ER(ER_MTS_FEATURE_IS_NOT_SUPPORTED),
"Temporary failed transaction retry",
"Such failure will force the slave to stop.");
}
=== modified file 'sql/share/errmsg-utf8.txt'
--- a/sql/share/errmsg-utf8.txt 2011-06-09 15:27:47 +0000
+++ b/sql/share/errmsg-utf8.txt 2011-06-12 19:05:05 +0000
@@ -6506,7 +6506,9 @@ ER_WARNING_NOT_COMPLETE_ROLLBACK_WITH_DR
ER_INDEX_COLUMN_TOO_LONG
eng "Index column size too large. The maximum column size is %lu bytes."
-ER_NO_FEATURE_ON_PARALLEL_SLAVE
+ER_MTS_FEATURE_IS_NOT_SUPPORTED
eng "%s is not supported in Parallel Slave. %s"
-ER_UPDATED_DBS_GREATER_MAX
+ER_MTS_UPDATED_DBS_GREATER_MAX
eng "Modified database names number exceeds the maximum %d; the names are not written into the replication event."
+ER_MTS_CANT_PARALLEL
+ eng "Can't execute the current event group in parallel mode running into event %s, relay-log name %s, position %s."
=== modified file 'sql/sp.cc'
--- a/sql/sp.cc 2011-06-09 15:27:47 +0000
+++ b/sql/sp.cc 2011-06-12 19:05:05 +0000
@@ -1150,7 +1150,7 @@ sp_create_routine(THD *thd, int type, sp
}
/* restore sql_mode when binloging */
thd->variables.sql_mode= saved_mode;
- thd->add_one_db_to_binlog_updated_dbs(sp->m_db.str);
+ thd->add_to_binlog_accessed_dbs(sp->m_db.str);
/* Such a statement can always go directly to binlog, no trans cache */
if (thd->binlog_query(THD::STMT_QUERY_TYPE,
log_query.c_ptr(), log_query.length(),
@@ -1224,7 +1224,7 @@ sp_drop_routine(THD *thd, int type, sp_n
if (ret == SP_OK)
{
- thd->add_one_db_to_binlog_updated_dbs(name->m_db.str);
+ thd->add_to_binlog_accessed_dbs(name->m_db.str);
if (write_bin_log(thd, TRUE, thd->query(), thd->query_length()))
ret= SP_INTERNAL_ERROR;
sp_cache_invalidate();
=== modified file 'sql/sql_acl.cc'
--- a/sql/sql_acl.cc 2011-05-31 13:52:09 +0000
+++ b/sql/sql_acl.cc 2011-06-12 19:05:05 +0000
@@ -3689,6 +3689,7 @@ int mysql_table_grant(THD *thd, TABLE_LI
should_write_to_binlog= TRUE;
db_name= table_list->get_db_name();
+ thd->add_to_binlog_accessed_dbs(db_name); // collecting db:s for MTS
table_name= table_list->get_table_name();
/* Find/create cached table grant */
@@ -3912,8 +3913,9 @@ bool mysql_routine_grant(THD *thd, TABLE
}
db_name= table_list->db;
+ if (write_to_binlog)
+ thd->add_to_binlog_accessed_dbs(db_name);
table_name= table_list->table_name;
-
grant_name= routine_hash_search(Str->host.str, NullS, db_name,
Str->user.str, table_name, is_proc, 1);
if (!grant_name)
@@ -4100,6 +4102,7 @@ bool mysql_grant(THD *thd, const char *d
my_error(ER_WRONG_USAGE, MYF(0), "DB GRANT", "GLOBAL PRIVILEGES");
result= -1;
}
+ thd->add_to_binlog_accessed_dbs(db);
}
else if (is_proxy)
{
=== modified file 'sql/sql_class.h'
--- a/sql/sql_class.h 2011-06-09 15:27:47 +0000
+++ b/sql/sql_class.h 2011-06-12 19:05:05 +0000
@@ -1951,31 +1951,13 @@ public:
}
/*
- MTS: initializer of binlog_accessed_db_names list
- */
- void set_binlog_accessed_db_names(List<char>* arg)
- {
- binlog_accessed_db_names= arg;
- }
-
- /*
MTS: resetter of binlog_accessed_db_names list normally
at the end of the query execution
*/
void clear_binlog_accessed_db_names() { binlog_accessed_db_names= NULL; }
/* MTS: method inserts a new unique name into binlog_updated_dbs */
- void add_to_binlog_updated_dbs(const char *db);
-
- /*
- MTS: method shortcuts initialization and insertion of just one db name
- into binlog_updated_dbs
- */
- void add_one_db_to_binlog_updated_dbs(const char *db)
- {
- set_binlog_accessed_db_names(new List<char>);
- binlog_accessed_db_names->push_back(strdup_root(mem_root, db));
- }
+ void add_to_binlog_accessed_dbs(const char *db);
#endif /* MYSQL_CLIENT */
=== modified file 'sql/sql_db.cc'
--- a/sql/sql_db.cc 2011-06-09 15:27:47 +0000
+++ b/sql/sql_db.cc 2011-06-12 19:05:05 +0000
@@ -657,7 +657,7 @@ not_silent:
*/
qinfo.db = db;
qinfo.db_len = strlen(db);
- thd->add_one_db_to_binlog_updated_dbs(db);
+ thd->add_to_binlog_accessed_dbs(db);
/*
These DDL methods and logging are protected with the exclusive
metadata lock on the schema
@@ -967,7 +967,7 @@ update_binlog:
if (query_pos != query_data_start)
{
- thd->add_one_db_to_binlog_updated_dbs(db);
+ thd->add_to_binlog_accessed_dbs(db);
/*
These DDL methods and logging are protected with the exclusive
metadata lock on the schema.
=== modified file 'sql/sql_rename.cc'
--- a/sql/sql_rename.cc 2011-06-09 15:27:47 +0000
+++ b/sql/sql_rename.cc 2011-06-12 19:05:05 +0000
@@ -317,12 +317,8 @@ do_rename(THD *thd, TABLE_LIST *ren_tabl
break;
}
- if (!thd->get_binlog_accessed_db_names())
- {
- thd->set_binlog_accessed_db_names(new List<char>);
- }
- thd->add_to_binlog_updated_dbs(ren_table->db);
- thd->add_to_binlog_updated_dbs(new_db);
+ thd->add_to_binlog_accessed_dbs(ren_table->db);
+ thd->add_to_binlog_accessed_dbs(new_db);
if (rc && !skip_error)
DBUG_RETURN(1);
=== modified file 'sql/sql_table.cc'
--- a/sql/sql_table.cc 2011-06-09 15:27:47 +0000
+++ b/sql/sql_table.cc 2011-06-12 19:05:05 +0000
@@ -2241,12 +2241,7 @@ int mysql_rm_table_no_locks(THD *thd, TA
find_temporary_table(thd, table) &&
table->mdl_request.ticket != NULL));
- /* MTS: similarly to decide_logging_format() gathering of the db names */
- if (!thd->get_binlog_accessed_db_names())
- {
- thd->set_binlog_accessed_db_names(new List<char>);
- }
- thd->add_to_binlog_updated_dbs(table->db);
+ thd->add_to_binlog_accessed_dbs(table->db);
/*
drop_temporary_table may return one of the following error codes:
@@ -4588,7 +4583,7 @@ bool mysql_create_table(THD *thd, TABLE_
(thd->is_current_stmt_binlog_format_row() &&
!(create_info->options & HA_LEX_CREATE_TMP_TABLE)))
{
- thd->add_one_db_to_binlog_updated_dbs(create_table->db);
+ thd->add_to_binlog_accessed_dbs(create_table->db);
result= write_bin_log(thd, TRUE, thd->query(), thd->query_length(), is_trans);
}
}
@@ -6002,13 +5997,9 @@ bool mysql_alter_table(THD *thd,char *ne
if (!new_db || !my_strcasecmp(table_alias_charset, new_db, db))
new_db= db;
- if (!thd->get_binlog_accessed_db_names())
- {
- thd->set_binlog_accessed_db_names(new List<char>);
- }
- thd->add_to_binlog_updated_dbs(db);
+ thd->add_to_binlog_accessed_dbs(db);
if (new_db != db)
- thd->add_to_binlog_updated_dbs(new_db);
+ thd->add_to_binlog_accessed_dbs(new_db);
build_table_filename(reg_path, sizeof(reg_path) - 1, db, table_name, reg_ext, 0);
build_table_filename(path, sizeof(path) - 1, db, table_name, "", 0);
=== modified file 'sql/sql_trigger.cc'
--- a/sql/sql_trigger.cc 2011-06-09 15:27:47 +0000
+++ b/sql/sql_trigger.cc 2011-06-12 19:05:05 +0000
@@ -528,7 +528,7 @@ end:
if (!result)
{
if (tables)
- thd->add_one_db_to_binlog_updated_dbs(tables->db);
+ thd->add_to_binlog_accessed_dbs(tables->db);
result= write_bin_log(thd, TRUE, stmt_query.ptr(), stmt_query.length());
}
=== modified file 'sql/sql_view.cc'
--- a/sql/sql_view.cc 2011-06-09 15:27:47 +0000
+++ b/sql/sql_view.cc 2011-06-12 19:05:05 +0000
@@ -690,7 +690,7 @@ bool mysql_create_view(THD *thd, TABLE_L
buff.append(views->source.str, views->source.length);
int errcode= query_error_code(thd, TRUE);
- thd->add_one_db_to_binlog_updated_dbs(views->db);
+ thd->add_to_binlog_accessed_dbs(views->db);
if (thd->binlog_query(THD::STMT_QUERY_TYPE,
buff.ptr(), buff.length(), FALSE, FALSE, FALSE, errcode))
res= TRUE;
@@ -1698,11 +1698,7 @@ bool mysql_drop_view(THD *thd, TABLE_LIS
}
continue;
}
- if (!thd->get_binlog_accessed_db_names())
- {
- thd->set_binlog_accessed_db_names(new List<char>);
- }
- thd->add_to_binlog_updated_dbs(view->db);
+ thd->add_to_binlog_accessed_dbs(view->db);
if (mysql_file_delete(key_file_frm, path, MYF(MY_WME)))
error= TRUE;
=== modified file 'storage/blackhole/ha_blackhole.cc'
--- a/storage/blackhole/ha_blackhole.cc 2011-05-26 15:20:09 +0000
+++ b/storage/blackhole/ha_blackhole.cc 2011-06-12 19:05:05 +0000
@@ -19,7 +19,7 @@
#include "unireg.h"
#include "probes_mysql.h"
#include "ha_blackhole.h"
-#include "sql_class.h" // THD, SYSTEM_THREAD_SLAVE_SQL
+#include "sql_class.h" // THD, SYSTEM_THREAD_SLAVE_*
/* Static declarations for handlerton */
@@ -114,7 +114,8 @@ int ha_blackhole::update_row(const uchar
{
DBUG_ENTER("ha_blackhole::update_row");
THD *thd= ha_thd();
- if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query() == NULL)
+ if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL ||
+ thd->system_thread == SYSTEM_THREAD_SLAVE_WORKER) && thd->query() == NULL)
DBUG_RETURN(0);
DBUG_RETURN(HA_ERR_WRONG_COMMAND);
}
@@ -123,7 +124,8 @@ int ha_blackhole::delete_row(const uchar
{
DBUG_ENTER("ha_blackhole::delete_row");
THD *thd= ha_thd();
- if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query() == NULL)
+ if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL ||
+ thd->system_thread == SYSTEM_THREAD_SLAVE_WORKER) && thd->query() == NULL)
DBUG_RETURN(0);
DBUG_RETURN(HA_ERR_WRONG_COMMAND);
}
@@ -142,7 +144,8 @@ int ha_blackhole::rnd_next(uchar *buf)
MYSQL_READ_ROW_START(table_share->db.str, table_share->table_name.str,
TRUE);
THD *thd= ha_thd();
- if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query() == NULL)
+ if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL ||
+ thd->system_thread == SYSTEM_THREAD_SLAVE_WORKER) && thd->query() == NULL)
rc= 0;
else
rc= HA_ERR_END_OF_FILE;
@@ -232,7 +235,8 @@ int ha_blackhole::index_read_map(uchar *
DBUG_ENTER("ha_blackhole::index_read");
MYSQL_INDEX_READ_ROW_START(table_share->db.str, table_share->table_name.str);
THD *thd= ha_thd();
- if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query() == NULL)
+ if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL ||
+ thd->system_thread == SYSTEM_THREAD_SLAVE_WORKER) && thd->query() == NULL)
rc= 0;
else
rc= HA_ERR_END_OF_FILE;
@@ -249,7 +253,8 @@ int ha_blackhole::index_read_idx_map(uch
DBUG_ENTER("ha_blackhole::index_read_idx");
MYSQL_INDEX_READ_ROW_START(table_share->db.str, table_share->table_name.str);
THD *thd= ha_thd();
- if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query() == NULL)
+ if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL ||
+ thd->system_thread == SYSTEM_THREAD_SLAVE_WORKER) && thd->query() == NULL)
rc= 0;
else
rc= HA_ERR_END_OF_FILE;
@@ -265,7 +270,8 @@ int ha_blackhole::index_read_last_map(uc
DBUG_ENTER("ha_blackhole::index_read_last");
MYSQL_INDEX_READ_ROW_START(table_share->db.str, table_share->table_name.str);
THD *thd= ha_thd();
- if (thd->system_thread == SYSTEM_THREAD_SLAVE_SQL && thd->query() == NULL)
+ if ((thd->system_thread == SYSTEM_THREAD_SLAVE_SQL ||
+ thd->system_thread == SYSTEM_THREAD_SLAVE_WORKER) && thd->query() == NULL)
rc= 0;
else
rc= HA_ERR_END_OF_FILE;
Attachment: [text/bzr-bundle] bzr/andrei.elkin@oracle.com-20110612193332-jb8qg7lf6z4ntr0f.bundle
| Thread |
|---|
| • bzr push into mysql-next-mr-wl5569 branch (andrei.elkin:3284 to 3286) WL#5569 | Andrei Elkin | 13 Jun |
