summaryrefslogtreecommitdiff
path: root/sql/log_event.cc
diff options
context:
space:
mode:
Diffstat (limited to 'sql/log_event.cc')
-rw-r--r--sql/log_event.cc155
1 files changed, 90 insertions, 65 deletions
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 28ac34c8a0f..70f0e6c2623 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -53,7 +53,6 @@
#include "rpl_constants.h"
#include "sql_digest.h"
#include "zlib.h"
-#include "my_atomic.h"
#define my_b_write_string(A, B) my_b_write((A), (uchar*)(B), (uint) (sizeof(B) - 1))
@@ -101,16 +100,11 @@ TYPELIB binlog_checksum_typelib=
TODO: correct the constant when it has been determined
(which main tree to push and when)
*/
-const uchar checksum_version_split_mysql[3]= {5, 6, 1};
-const ulong checksum_version_product_mysql=
- (checksum_version_split_mysql[0] * 256 +
- checksum_version_split_mysql[1]) * 256 +
- checksum_version_split_mysql[2];
-const uchar checksum_version_split_mariadb[3]= {5, 3, 0};
-const ulong checksum_version_product_mariadb=
- (checksum_version_split_mariadb[0] * 256 +
- checksum_version_split_mariadb[1]) * 256 +
- checksum_version_split_mariadb[2];
+const Version checksum_version_split_mysql(5, 6, 1);
+const Version checksum_version_split_mariadb(5, 3, 0);
+
+// First MySQL version with fraction seconds
+const Version fsp_version_split_mysql(5, 6, 0);
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
static int rows_event_stmt_cleanup(rpl_group_info *rgi, THD* thd);
@@ -2753,9 +2747,7 @@ log_event_print_value(IO_CACHE *file, PRINT_EVENT_INFO *print_event_info,
goto return_null;
uint bin_size= my_decimal_get_binary_size(precision, decimals);
- my_decimal dec;
- binary2my_decimal(E_DEC_FATAL_ERROR, (uchar*) ptr, &dec,
- precision, decimals);
+ my_decimal dec((const uchar *) ptr, precision, decimals);
int length= DECIMAL_MAX_STR_LENGTH;
char buff[DECIMAL_MAX_STR_LENGTH + 1];
decimal2string(&dec, buff, &length, 0, 0, 0);
@@ -4426,7 +4418,7 @@ Query_log_event::Query_log_event(THD* thd_arg, const char* query_arg, size_t que
have to use the transactional cache to ensure we don't
calculate any checksum for the CREATE part.
*/
- trx_cache= (lex->select_lex.item_list.elements &&
+ trx_cache= (lex->first_select_lex()->item_list.elements &&
thd->is_current_stmt_binlog_format_row()) ||
(thd->variables.option_bits & OPTION_GTID_BEGIN);
use_cache= (lex->tmp_table() &&
@@ -4558,6 +4550,7 @@ code_name(int code)
}
#endif
+
/**
Macro to check that there is enough space to read from memory.
@@ -4781,6 +4774,30 @@ Query_log_event::Query_log_event(const char* buf, uint event_len,
}
}
+#if !defined(MYSQL_CLIENT)
+ if (description_event->server_version_split.kind ==
+ Format_description_log_event::master_version_split::KIND_MYSQL)
+ {
+ // Handle MariaDB/MySQL incompatible sql_mode bits
+ sql_mode_t mysql_sql_mode= sql_mode;
+ sql_mode&= MODE_MASK_MYSQL_COMPATIBLE; // Unset MySQL specific bits
+
+ /*
+ sql_mode flags related to fraction second rounding/truncation
+ have opposite meaning in MySQL vs MariaDB.
+ MySQL:
+ - rounds fractional seconds by default
+ - truncates if TIME_TRUNCATE_FRACTIONAL is set
+ MariaDB:
+ - truncates fractional seconds by default
+ - rounds if TIME_ROUND_FRACTIONAL is set
+ */
+ if (description_event->server_version_split >= fsp_version_split_mysql &&
+ !(mysql_sql_mode & MODE_MYSQL80_TIME_TRUNCATE_FRACTIONAL))
+ sql_mode|= MODE_TIME_ROUND_FRACTIONAL;
+ }
+#endif
+
/**
Layout for the data buffer is as follows
+--------+-----------+------+------+---------+----+-------+
@@ -5614,7 +5631,7 @@ int Query_log_event::do_apply_event(rpl_group_info *rgi,
gtid= rgi->current_gtid;
if (unlikely(rpl_global_gtid_slave_state->record_gtid(thd, &gtid,
sub_id,
- rgi, false,
+ true, false,
&hton)))
{
int errcode= thd->get_stmt_da()->sql_errno();
@@ -5794,6 +5811,14 @@ compare_errors:
"unexpected success or fatal error"),
thd->get_db(), query_arg);
thd->is_slave_error= 1;
+#ifdef WITH_WSREP
+ if (thd->wsrep_apply_toi && wsrep_must_ignore_error(thd))
+ {
+ thd->clear_error(1);
+ thd->killed= NOT_KILLED;
+ thd->wsrep_has_ignored_error= true;
+ }
+#endif /* WITH_WSREP */
}
/*
@@ -6589,26 +6614,24 @@ bool Format_description_log_event::start_decryption(Start_encryption_log_event*
return crypto_data.init(sele->crypto_scheme, sele->key_version);
}
-static inline void
-do_server_version_split(char* version,
- Format_description_log_event::master_version_split *split_versions)
+
+Version::Version(const char *version, const char **endptr)
{
- char *p= version, *r;
+ const char *p= version;
ulong number;
for (uint i= 0; i<=2; i++)
{
+ char *r;
number= strtoul(p, &r, 10);
/*
It is an invalid version if any version number greater than 255 or
first number is not followed by '.'.
*/
if (number < 256 && (*r == '.' || i != 0))
- split_versions->ver[i]= (uchar) number;
+ m_ver[i]= (uchar) number;
else
{
- split_versions->ver[0]= 0;
- split_versions->ver[1]= 0;
- split_versions->ver[2]= 0;
+ *this= Version();
break;
}
@@ -6616,12 +6639,19 @@ do_server_version_split(char* version,
if (*r == '.')
p++; // skip the dot
}
+ endptr[0]= p;
+}
+
+
+Format_description_log_event::
+ master_version_split::master_version_split(const char *version)
+{
+ const char *p;
+ static_cast<Version*>(this)[0]= Version(version, &p);
if (strstr(p, "MariaDB") != 0 || strstr(p, "-maria-") != 0)
- split_versions->kind=
- Format_description_log_event::master_version_split::KIND_MARIADB;
+ kind= KIND_MARIADB;
else
- split_versions->kind=
- Format_description_log_event::master_version_split::KIND_MYSQL;
+ kind= KIND_MYSQL;
}
@@ -6635,20 +6665,14 @@ do_server_version_split(char* version,
*/
void Format_description_log_event::calc_server_version_split()
{
- do_server_version_split(server_version, &server_version_split);
+ server_version_split= master_version_split(server_version);
DBUG_PRINT("info",("Format_description_log_event::server_version_split:"
" '%s' %d %d %d", server_version,
- server_version_split.ver[0],
- server_version_split.ver[1], server_version_split.ver[2]));
+ server_version_split[0],
+ server_version_split[1], server_version_split[2]));
}
-static inline ulong
-version_product(const Format_description_log_event::master_version_split* version_split)
-{
- return ((version_split->ver[0] * 256 + version_split->ver[1]) * 256
- + version_split->ver[2]);
-}
/**
@return TRUE is the event's version is earlier than one that introduced
@@ -6658,9 +6682,9 @@ bool
Format_description_log_event::is_version_before_checksum(const master_version_split
*version_split)
{
- return version_product(version_split) <
+ return *version_split <
(version_split->kind == master_version_split::KIND_MARIADB ?
- checksum_version_product_mariadb : checksum_version_product_mysql);
+ checksum_version_split_mariadb : checksum_version_split_mysql);
}
/**
@@ -6676,7 +6700,6 @@ enum enum_binlog_checksum_alg get_checksum_alg(const char* buf, ulong len)
{
enum enum_binlog_checksum_alg ret;
char version[ST_SERVER_VER_LEN];
- Format_description_log_event::master_version_split version_split;
DBUG_ENTER("get_checksum_alg");
DBUG_ASSERT(buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT);
@@ -6686,7 +6709,7 @@ enum enum_binlog_checksum_alg get_checksum_alg(const char* buf, ulong len)
ST_SERVER_VER_LEN);
version[ST_SERVER_VER_LEN - 1]= 0;
- do_server_version_split(version, &version_split);
+ Format_description_log_event::master_version_split version_split(version);
ret= Format_description_log_event::is_version_before_checksum(&version_split)
? BINLOG_CHECKSUM_ALG_UNDEF
: (enum_binlog_checksum_alg)buf[len - BINLOG_CHECKSUM_LEN - BINLOG_CHECKSUM_ALG_DESC_LEN];
@@ -7437,8 +7460,9 @@ int Load_log_event::do_apply_event(NET* net, rpl_group_info *rgi,
ex.skip_lines = skip_lines;
List<Item> field_list;
- thd->lex->select_lex.context.resolve_in_table_list_only(&tables);
- set_fields(tables.db.str, field_list, &thd->lex->select_lex.context);
+ thd->lex->first_select_lex()->context.resolve_in_table_list_only(&tables);
+ set_fields(tables.db.str,
+ field_list, &thd->lex->first_select_lex()->context);
thd->variables.pseudo_thread_id= thread_id;
if (net)
{
@@ -8076,16 +8100,13 @@ Gtid_log_event::do_apply_event(rpl_group_info *rgi)
switch (flags2 & (FL_DDL | FL_TRANSACTIONAL))
{
case FL_TRANSACTIONAL:
- my_atomic_add64_explicit((volatile int64 *)&mi->total_trans_groups, 1,
- MY_MEMORY_ORDER_RELAXED);
+ mi->total_trans_groups++;
break;
case FL_DDL:
- my_atomic_add64_explicit((volatile int64 *)&mi->total_ddl_groups, 1,
- MY_MEMORY_ORDER_RELAXED);
+ mi->total_ddl_groups++;
break;
default:
- my_atomic_add64_explicit((volatile int64 *)&mi->total_non_trans_groups, 1,
- MY_MEMORY_ORDER_RELAXED);
+ mi->total_non_trans_groups++;
}
if (flags2 & FL_STANDALONE)
@@ -8417,7 +8438,7 @@ Gtid_list_log_event::do_apply_event(rpl_group_info *rgi)
{
if ((ret= rpl_global_gtid_slave_state->record_gtid(thd, &list[i],
sub_id_list[i],
- NULL, false, &hton)))
+ false, false, &hton)))
return ret;
rpl_global_gtid_slave_state->update_state_hash(sub_id_list[i], &list[i],
hton, NULL);
@@ -8954,7 +8975,7 @@ int Xid_log_event::do_apply_event(rpl_group_info *rgi)
rgi->gtid_pending= false;
gtid= rgi->current_gtid;
- err= rpl_global_gtid_slave_state->record_gtid(thd, &gtid, sub_id, rgi,
+ err= rpl_global_gtid_slave_state->record_gtid(thd, &gtid, sub_id, true,
false, &hton);
if (unlikely(err))
{
@@ -9092,11 +9113,8 @@ void User_var_log_event::pack_info(Protocol* protocol)
String buf(buf_mem, sizeof(buf_mem), system_charset_info);
char buf2[DECIMAL_MAX_STR_LENGTH+1];
String str(buf2, sizeof(buf2), &my_charset_bin);
- my_decimal dec;
buf.length(0);
- binary2my_decimal(E_DEC_FATAL_ERROR, (uchar*) (val+2), &dec, val[0],
- val[1]);
- my_decimal2string(E_DEC_FATAL_ERROR, &dec, 0, 0, 0, &str);
+ my_decimal((const uchar *) (val + 2), val[0], val[1]).to_string(&str);
if (user_var_append_name_part(protocol->thd, &buf, name, name_len) ||
buf.append(buf2))
return;
@@ -11309,13 +11327,13 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
{
WSREP_WARN("BF applier failed to open_and_lock_tables: %u, fatal: %d "
"wsrep = (exec_mode: %d conflict_state: %d seqno: %lld)",
- thd->get_stmt_da()->sql_errno(),
- thd->is_fatal_error,
- thd->wsrep_exec_mode,
- thd->wsrep_conflict_state,
- (long long)wsrep_thd_trx_seqno(thd));
+ thd->get_stmt_da()->sql_errno(),
+ thd->is_fatal_error,
+ thd->wsrep_cs().mode(),
+ thd->wsrep_trx().state(),
+ (long long) wsrep_thd_trx_seqno(thd));
}
-#endif
+#endif /* WITH_WSREP */
if ((thd->is_slave_error || thd->is_fatal_error) &&
!is_parallel_retry_error(rgi, actual_error))
{
@@ -11452,10 +11470,10 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
#ifdef HAVE_QUERY_CACHE
#ifdef WITH_WSREP
/*
- Moved invalidation right before the call to rows_event_stmt_cleanup(),
- to avoid query cache being polluted with stale entries.
+ Moved invalidation right before the call to rows_event_stmt_cleanup(),
+ to avoid query cache being polluted with stale entries,
*/
- if (! (WSREP(thd) && (thd->wsrep_exec_mode == REPL_RECV)))
+ if (! (WSREP(thd) && wsrep_thd_is_applying(thd)))
{
#endif /* WITH_WSREP */
query_cache.invalidate_locked_for_write(thd, rgi->tables_to_lock);
@@ -11568,6 +11586,13 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
bool ignored_error= (idempotent_error == 0 ?
ignored_error_code(actual_error) : 0);
+#ifdef WITH_WSREP
+ if (WSREP(thd) && wsrep_ignored_error_code(this, actual_error))
+ {
+ idempotent_error= true;
+ thd->wsrep_has_ignored_error= true;
+ }
+#endif /* WITH_WSREP */
if (idempotent_error || ignored_error)
{
if (global_system_variables.log_warnings)
@@ -11655,7 +11680,7 @@ int Rows_log_event::do_apply_event(rpl_group_info *rgi)
restore_empty_query_table_list(thd->lex);
#if defined(WITH_WSREP) && defined(HAVE_QUERY_CACHE)
- if (WSREP(thd) && thd->wsrep_exec_mode == REPL_RECV)
+ if (WSREP(thd) && wsrep_thd_is_applying(thd))
{
query_cache.invalidate_locked_for_write(thd, rgi->tables_to_lock);
}
View Lorry Controller Status