Commit 76551969 authored by Sergei Petrunia's avatar Sergei Petrunia

Enable basic XA between MyRocks and the binlog

This allows basic master crash-safety
- Un-comment and update relevant parts of the code
- Make rocksdb_rpl suite work like other MyRocks testsuites
  (load the MyRocks plugin, don't start if it is not compiled in, etc)
- For now, disable all tests in the rocksdb_rpl suite.
- MariaDB-fication of rpl_rocksdb_2p_crash_recover test.
parent a544225d
...@@ -170,7 +170,6 @@ static std::shared_ptr<Rdb_tbl_prop_coll_factory> properties_collector_factory; ...@@ -170,7 +170,6 @@ static std::shared_ptr<Rdb_tbl_prop_coll_factory> properties_collector_factory;
Rdb_dict_manager dict_manager; Rdb_dict_manager dict_manager;
Rdb_cf_manager cf_manager; Rdb_cf_manager cf_manager;
Rdb_ddl_manager ddl_manager; Rdb_ddl_manager ddl_manager;
const char *m_mysql_gtid;
Rdb_binlog_manager binlog_manager; Rdb_binlog_manager binlog_manager;
#ifndef _WIN32 #ifndef _WIN32
...@@ -1668,8 +1667,11 @@ class Rdb_transaction { ...@@ -1668,8 +1667,11 @@ class Rdb_transaction {
public: public:
const char *m_mysql_log_file_name; const char *m_mysql_log_file_name;
my_off_t m_mysql_log_offset; my_off_t m_mysql_log_offset;
#ifdef MARIAROCKS_NOT_YET
// TODO: MariaDB probably doesn't need these at all:
const char *m_mysql_gtid; const char *m_mysql_gtid;
const char *m_mysql_max_gtid; const char *m_mysql_max_gtid;
#endif
String m_detailed_error; String m_detailed_error;
int64_t m_snapshot_timestamp = 0; int64_t m_snapshot_timestamp = 0;
bool m_ddl_transaction; bool m_ddl_transaction;
...@@ -1856,13 +1858,10 @@ class Rdb_transaction { ...@@ -1856,13 +1858,10 @@ class Rdb_transaction {
rollback(); rollback();
return true; return true;
} else { } else {
#ifdef MARIAROCKS_NOT_YET mysql_bin_log_commit_pos(m_thd, &m_mysql_log_offset,
my_core::thd_binlog_pos(m_thd, &m_mysql_log_file_name, &m_mysql_log_file_name);
&m_mysql_log_offset, &m_mysql_gtid,
&m_mysql_max_gtid);
binlog_manager.update(m_mysql_log_file_name, m_mysql_log_offset, binlog_manager.update(m_mysql_log_file_name, m_mysql_log_offset,
m_mysql_max_gtid, get_write_batch()); get_write_batch());
#endif
return commit_no_binlog(); return commit_no_binlog();
} }
} }
...@@ -2683,26 +2682,25 @@ static bool rocksdb_flush_wal(handlerton* hton __attribute__((__unused__))) ...@@ -2683,26 +2682,25 @@ static bool rocksdb_flush_wal(handlerton* hton __attribute__((__unused__)))
*/ */
static int rocksdb_prepare(handlerton* hton, THD* thd, bool prepare_tx) static int rocksdb_prepare(handlerton* hton, THD* thd, bool prepare_tx)
{ {
#ifdef MARIAROCKS_NOT_YET
// This is "ASYNC_COMMIT" feature which is only in webscalesql // This is "ASYNC_COMMIT" feature which is only in webscalesql
bool async=false; bool async=false;
#endif
Rdb_transaction *&tx = get_tx_from_thd(thd); Rdb_transaction *&tx = get_tx_from_thd(thd);
if (!tx->can_prepare()) { if (!tx->can_prepare()) {
return HA_EXIT_FAILURE; return HA_EXIT_FAILURE;
} }
#ifdef MARIAROCKS_NOT_YET // disable prepare/commit
if (prepare_tx || if (prepare_tx ||
(!my_core::thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) { (!my_core::thd_test_options(thd, OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN))) {
/* We were instructed to prepare the whole transaction, or /* We were instructed to prepare the whole transaction, or
this is an SQL statement end and autocommit is on */ this is an SQL statement end and autocommit is on */
#ifdef MARIAROCKS_NOT_YET // disable prepare/commit
std::vector<st_slave_gtid_info> slave_gtid_info; std::vector<st_slave_gtid_info> slave_gtid_info;
my_core::thd_slave_gtid_info(thd, &slave_gtid_info); my_core::thd_slave_gtid_info(thd, &slave_gtid_info);
for (const auto &it : slave_gtid_info) { for (const auto &it : slave_gtid_info) {
rocksdb::WriteBatchBase *const write_batch = tx->get_blind_write_batch(); rocksdb::WriteBatchBase *const write_batch = tx->get_blind_write_batch();
binlog_manager.update_slave_gtid_info(it.id, it.db, it.gtid, write_batch); binlog_manager.update_slave_gtid_info(it.id, it.db, it.gtid, write_batch);
} }
#endif
if (tx->is_two_phase()) { if (tx->is_two_phase()) {
if (thd->durability_property == HA_IGNORE_DURABILITY || async) { if (thd->durability_property == HA_IGNORE_DURABILITY || async) {
...@@ -2713,10 +2711,10 @@ static int rocksdb_prepare(handlerton* hton, THD* thd, bool prepare_tx) ...@@ -2713,10 +2711,10 @@ static int rocksdb_prepare(handlerton* hton, THD* thd, bool prepare_tx)
if (!tx->prepare(rdb_xid_to_string(xid))) { if (!tx->prepare(rdb_xid_to_string(xid))) {
return HA_EXIT_FAILURE; return HA_EXIT_FAILURE;
} }
if (thd->durability_property == HA_IGNORE_DURABILITY if (thd->durability_property == HA_IGNORE_DURABILITY )
#ifdef MARIAROCKS_NOT_YET #ifdef MARIAROCKS_NOT_YET
&& &&
THDVAR(thd, flush_log_at_trx_commit)) { THDVAR(thd, flush_log_at_trx_commit))
#endif #endif
{ {
#ifdef MARIAROCKS_NOT_YET #ifdef MARIAROCKS_NOT_YET
...@@ -2733,7 +2731,6 @@ static int rocksdb_prepare(handlerton* hton, THD* thd, bool prepare_tx) ...@@ -2733,7 +2731,6 @@ static int rocksdb_prepare(handlerton* hton, THD* thd, bool prepare_tx)
DEBUG_SYNC(thd, "rocksdb.prepared"); DEBUG_SYNC(thd, "rocksdb.prepared");
} }
#endif
return HA_EXIT_SUCCESS; return HA_EXIT_SUCCESS;
} }
......
if (`SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.ENGINES WHERE engine = 'rocksdb' AND support IN ('YES', 'DEFAULT', 'ENABLED')`)
{
--skip Test requires engine RocksDB.
}
--disable_query_log
# Table statistics can vary depending on when the memtables are flushed, so
# flush them at the beginning of the test to ensure the test runs consistently.
set global rocksdb_force_flush_memtable_now = true;
--enable_query_log
--loose-enable-rocksdb
--loose-enable-rocksdb_global_info
--loose-enable-rocksdb_ddl
--loose-enable-rocksdb_cf_options
--loose-enable_rocksdb_perf_context
--loose-enable_rocksdb_perf_context_global
--loose-enable-rocksdb_index_file_map
--loose-enable-rocksdb_dbstats
--loose-enable-rocksdb_cfstats
--loose-enable-rocksdb_lock_info
--loose-enable-rocksdb_trx
--loose-enable-rocksdb_locks
!include include/default_my.cnf
[server]
skip-innodb
default-storage-engine=rocksdb
sql-mode=NO_ENGINE_SUBSTITUTION
explicit-defaults-for-timestamp=1
loose-rocksdb_lock_wait_timeout=1
loose-rocksdb_strict_collation_check=0
loose-rocksdb-flush-log-at-trx-commit=0
# The following is to get rid of the harmless
# "Deadlock found when trying to get lock" errors, see MDEV-12285.
log-warnings=1
DROP TABLE IF EXISTS t1; DROP TABLE IF EXISTS t1;
create table t1 (a int primary key, msg varchar(255)) engine=rocksdb; create table t1 (a int primary key, msg varchar(255)) engine=rocksdb;
SET GLOBAL ROCKSDB_ENABLE_2PC = ON; SET GLOBAL ROCKSDB_ENABLE_2PC = ON;
SET SESSION debug="d,crash_commit_after_prepare"; SET SESSION debug_dbug="d,crash_commit_after_prepare";
insert into t1 values (1, 'dogz'); insert into t1 values (1, 'dogz');
select * from t1; select * from t1;
a msg a msg
SET GLOBAL ROCKSDB_ENABLE_2PC = ON; SET GLOBAL ROCKSDB_ENABLE_2PC = ON;
SET SESSION debug="d,crash_commit_after_log"; SET SESSION debug_dbug="d,crash_commit_after_log";
insert into t1 values (2, 'catz'), (3, 'men'); insert into t1 values (2, 'catz'), (3, 'men');
select * from t1; select * from t1;
a msg a msg
2 catz 2 catz
3 men 3 men
SET GLOBAL ROCKSDB_ENABLE_2PC = ON; SET GLOBAL ROCKSDB_ENABLE_2PC = ON;
SET SESSION debug="d,crash_commit_after"; SET SESSION debug_dbug="d,crash_commit_after";
insert into t1 values (4, 'cars'), (5, 'foo'); insert into t1 values (4, 'cars'), (5, 'foo');
select * from t1; select * from t1;
a msg a msg
...@@ -22,7 +22,7 @@ a msg ...@@ -22,7 +22,7 @@ a msg
4 cars 4 cars
5 foo 5 foo
SET GLOBAL ROCKSDB_ENABLE_2PC = OFF; SET GLOBAL ROCKSDB_ENABLE_2PC = OFF;
SET SESSION debug="d,crash_commit_after_log"; SET SESSION debug_dbug="d,crash_commit_after_log";
insert into t1 values (6, 'shipz'), (7, 'tankz'); insert into t1 values (6, 'shipz'), (7, 'tankz');
select * from t1; select * from t1;
a msg a msg
...@@ -31,7 +31,7 @@ a msg ...@@ -31,7 +31,7 @@ a msg
4 cars 4 cars
5 foo 5 foo
SET GLOBAL ROCKSDB_ENABLE_2PC = OFF; SET GLOBAL ROCKSDB_ENABLE_2PC = OFF;
SET SESSION debug="d,crash_commit_after"; SET SESSION debug_dbug="d,crash_commit_after";
insert into t1 values (8, 'space'), (9, 'time'); insert into t1 values (8, 'space'), (9, 'time');
select * from t1; select * from t1;
a msg a msg
......
--ignore-db-dirs=.rocksdb --plugin-load=$HA_ROCKSDB_SO --default-storage-engine=rocksdb
package My::Suite::Rocksdb;
#
# Note: ../rocksdb_sys_vars/suite.pm file has a similar
# function. If you modify this file, consider modifying that one, too.
#
@ISA = qw(My::Suite);
use My::Find;
use File::Basename;
use strict;
sub is_default { not $::opt_embedded_server }
my $sst_dump=
::mtr_exe_maybe_exists(
"$::bindir/storage/rocksdb$::opt_vs_config/sst_dump",
"$::path_client_bindir/sst_dump");
return "RocksDB is not compiled, no sst_dump" unless $sst_dump;
$ENV{MARIAROCKS_SST_DUMP}="$sst_dump";
# Temporarily disable testing under valgrind, due to MDEV-12439
return "RocksDB tests disabled under valgrind" if ($::opt_valgrind);
bless { };
# rpl_rocksdb_2pc_crash_recover
consistent_snapshot_mixed_engines : Didn't try with MariaDB, yet
multiclient_2pc : Didn't try with MariaDB, yet
rpl_crash_safe_wal_corrupt : Didn't try with MariaDB, yet
rpl_gtid_crash_safe : Didn't try with MariaDB, yet
rpl_gtid_crash_safe_wal_corrupt : Didn't try with MariaDB, yet
rpl_gtid_rocksdb_sys_header : Didn't try with MariaDB, yet
rpl_no_unique_check_on_lag : Didn't try with MariaDB, yet
rpl_no_unique_check_on_lag_mts : Didn't try with MariaDB, yet
rpl_rocksdb_snapshot : Didn't try with MariaDB, yet
rpl_rocksdb_snapshot_without_gtid : Didn't try with MariaDB, yet
rpl_rocksdb_stress_crash : Didn't try with MariaDB, yet
rpl_skip_trx_api_binlog_format : Didn't try with MariaDB, yet
singledelete_idempotent_recovery : Didn't try with MariaDB, yet
singledelete_idempotent_table : Didn't try with MariaDB, yet
--gtid_mode=ON --enforce_gtid_consistency --log_bin --log_slave_updates --rocksdb_flush_log_at_trx_commit=1 --rocksdb_write_disable_wal=OFF --log_bin --log_slave_updates --rocksdb_flush_log_at_trx_commit=1
--gtid_mode=ON --enforce_gtid_consistency --log_bin --log_slave_updates --log_bin --log_slave_updates
...@@ -10,7 +10,7 @@ create table t1 (a int primary key, msg varchar(255)) engine=rocksdb; ...@@ -10,7 +10,7 @@ create table t1 (a int primary key, msg varchar(255)) engine=rocksdb;
SET GLOBAL ROCKSDB_ENABLE_2PC = ON; SET GLOBAL ROCKSDB_ENABLE_2PC = ON;
--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
SET SESSION debug="d,crash_commit_after_prepare"; SET SESSION debug_dbug="d,crash_commit_after_prepare";
--error 0,2013 --error 0,2013
insert into t1 values (1, 'dogz'); insert into t1 values (1, 'dogz');
--enable_reconnect --enable_reconnect
...@@ -19,7 +19,7 @@ select * from t1; ...@@ -19,7 +19,7 @@ select * from t1;
SET GLOBAL ROCKSDB_ENABLE_2PC = ON; SET GLOBAL ROCKSDB_ENABLE_2PC = ON;
--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
SET SESSION debug="d,crash_commit_after_log"; SET SESSION debug_dbug="d,crash_commit_after_log";
--error 0,2013 --error 0,2013
insert into t1 values (2, 'catz'), (3, 'men'); insert into t1 values (2, 'catz'), (3, 'men');
--enable_reconnect --enable_reconnect
...@@ -28,7 +28,7 @@ select * from t1; ...@@ -28,7 +28,7 @@ select * from t1;
SET GLOBAL ROCKSDB_ENABLE_2PC = ON; SET GLOBAL ROCKSDB_ENABLE_2PC = ON;
--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
SET SESSION debug="d,crash_commit_after"; SET SESSION debug_dbug="d,crash_commit_after";
--error 0,2013 --error 0,2013
insert into t1 values (4, 'cars'), (5, 'foo'); insert into t1 values (4, 'cars'), (5, 'foo');
--enable_reconnect --enable_reconnect
...@@ -37,7 +37,7 @@ select * from t1; ...@@ -37,7 +37,7 @@ select * from t1;
SET GLOBAL ROCKSDB_ENABLE_2PC = OFF; SET GLOBAL ROCKSDB_ENABLE_2PC = OFF;
--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
SET SESSION debug="d,crash_commit_after_log"; SET SESSION debug_dbug="d,crash_commit_after_log";
--error 0,2013 --error 0,2013
insert into t1 values (6, 'shipz'), (7, 'tankz'); insert into t1 values (6, 'shipz'), (7, 'tankz');
--enable_reconnect --enable_reconnect
...@@ -46,7 +46,7 @@ select * from t1; ...@@ -46,7 +46,7 @@ select * from t1;
SET GLOBAL ROCKSDB_ENABLE_2PC = OFF; SET GLOBAL ROCKSDB_ENABLE_2PC = OFF;
--exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect --exec echo "restart" > $MYSQLTEST_VARDIR/tmp/mysqld.1.expect
SET SESSION debug="d,crash_commit_after"; SET SESSION debug_dbug="d,crash_commit_after";
--error 0,2013 --error 0,2013
insert into t1 values (8, 'space'), (9, 'time'); insert into t1 values (8, 'space'), (9, 'time');
--enable_reconnect --enable_reconnect
......
...@@ -3961,12 +3961,10 @@ void Rdb_binlog_manager::cleanup() {} ...@@ -3961,12 +3961,10 @@ void Rdb_binlog_manager::cleanup() {}
write succeeded or not is not possible here. write succeeded or not is not possible here.
@param binlog_name Binlog name @param binlog_name Binlog name
@param binlog_pos Binlog pos @param binlog_pos Binlog pos
@param binlog_gtid Binlog max GTID
@param batch WriteBatch @param batch WriteBatch
*/ */
void Rdb_binlog_manager::update(const char *const binlog_name, void Rdb_binlog_manager::update(const char *const binlog_name,
const my_off_t binlog_pos, const my_off_t binlog_pos,
const char *const binlog_max_gtid,
rocksdb::WriteBatchBase *const batch) { rocksdb::WriteBatchBase *const batch) {
if (binlog_name && binlog_pos) { if (binlog_name && binlog_pos) {
// max binlog length (512) + binlog pos (4) + binlog gtid (57) < 1024 // max binlog length (512) + binlog pos (4) + binlog gtid (57) < 1024
...@@ -3974,7 +3972,7 @@ void Rdb_binlog_manager::update(const char *const binlog_name, ...@@ -3974,7 +3972,7 @@ void Rdb_binlog_manager::update(const char *const binlog_name,
uchar value_buf[RDB_MAX_BINLOG_INFO_LEN]; uchar value_buf[RDB_MAX_BINLOG_INFO_LEN];
m_dict->put_key( m_dict->put_key(
batch, m_key_slice, batch, m_key_slice,
pack_value(value_buf, binlog_name, binlog_pos, binlog_max_gtid)); pack_value(value_buf, binlog_name, binlog_pos, NULL));
} }
} }
...@@ -4009,7 +4007,6 @@ bool Rdb_binlog_manager::read(char *const binlog_name, ...@@ -4009,7 +4007,6 @@ bool Rdb_binlog_manager::read(char *const binlog_name,
@param buf Preallocated buffer to set binlog info. @param buf Preallocated buffer to set binlog info.
@param binlog_name Binlog name @param binlog_name Binlog name
@param binlog_pos Binlog pos @param binlog_pos Binlog pos
@param binlog_gtid Binlog GTID
@return rocksdb::Slice converted from buf and its length @return rocksdb::Slice converted from buf and its length
*/ */
rocksdb::Slice rocksdb::Slice
...@@ -4038,15 +4035,21 @@ Rdb_binlog_manager::pack_value(uchar *const buf, const char *const binlog_name, ...@@ -4038,15 +4035,21 @@ Rdb_binlog_manager::pack_value(uchar *const buf, const char *const binlog_name,
// store binlog gtid length. // store binlog gtid length.
// If gtid was not set, store 0 instead // If gtid was not set, store 0 instead
#ifdef MARIAROCKS_NOT_YET
const uint16_t binlog_gtid_len = binlog_gtid ? (uint16_t)strlen(binlog_gtid) : 0; const uint16_t binlog_gtid_len = binlog_gtid ? (uint16_t)strlen(binlog_gtid) : 0;
rdb_netbuf_store_uint16(buf + pack_len, binlog_gtid_len); rdb_netbuf_store_uint16(buf + pack_len, binlog_gtid_len);
#endif
pack_len += sizeof(uint16); pack_len += sizeof(uint16);
// MariaDB:
rdb_netbuf_store_uint16(buf + pack_len, 0);
#ifdef MARIAROCKS_NOT_YET
if (binlog_gtid_len > 0) { if (binlog_gtid_len > 0) {
// store binlog gtid // store binlog gtid
memcpy(buf + pack_len, binlog_gtid, binlog_gtid_len); memcpy(buf + pack_len, binlog_gtid, binlog_gtid_len);
pack_len += binlog_gtid_len; pack_len += binlog_gtid_len;
} }
#endif
return rocksdb::Slice((char *)buf, pack_len); return rocksdb::Slice((char *)buf, pack_len);
} }
......
...@@ -1082,7 +1082,6 @@ class Rdb_binlog_manager { ...@@ -1082,7 +1082,6 @@ class Rdb_binlog_manager {
bool init(Rdb_dict_manager *const dict); bool init(Rdb_dict_manager *const dict);
void cleanup(); void cleanup();
void update(const char *const binlog_name, const my_off_t binlog_pos, void update(const char *const binlog_name, const my_off_t binlog_pos,
const char *const binlog_max_gtid,
rocksdb::WriteBatchBase *const batch); rocksdb::WriteBatchBase *const batch);
bool read(char *const binlog_name, my_off_t *const binlog_pos, bool read(char *const binlog_name, my_off_t *const binlog_pos,
char *const binlog_gtid) const; char *const binlog_gtid) const;
......
...@@ -49,4 +49,7 @@ typedef struct my_io_perf_struct my_io_perf_t; ...@@ -49,4 +49,7 @@ typedef struct my_io_perf_struct my_io_perf_t;
std::vector<std::string> split_into_vector(const std::string& input, std::vector<std::string> split_into_vector(const std::string& input,
char delimiter); char delimiter);
void
mysql_bin_log_commit_pos(THD *thd, ulonglong *out_pos, const char **out_file);
#endif #endif
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment