diff options
Diffstat (limited to '')
| -rw-r--r-- | sql/log_event.cc | 4182 |
1 files changed, 4182 insertions, 0 deletions
diff --git a/sql/log_event.cc b/sql/log_event.cc new file mode 100644 index 00000000..5e255646 --- /dev/null +++ b/sql/log_event.cc @@ -0,0 +1,4182 @@ +/* + Copyright (c) 2000, 2018, Oracle and/or its affiliates. + Copyright (c) 2009, 2022, MariaDB + + This program is free software; you can redistribute it and/or modify + it under the terms of the GNU General Public License as published by + the Free Software Foundation; version 2 of the License. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1335 USA */ + + +#include "mariadb.h" +#include "sql_priv.h" +#include "handler.h" +#ifndef MYSQL_CLIENT +#include "unireg.h" +#include "log_event.h" +#include "sql_base.h" // close_thread_tables +#include "sql_cache.h" // QUERY_CACHE_FLAGS_SIZE +#include "sql_locale.h" // MY_LOCALE, my_locale_by_number, my_locale_en_US +#include "key.h" // key_copy +#include "lock.h" // mysql_unlock_tables +#include "sql_parse.h" // mysql_test_parse_for_slave +#include "tztime.h" // struct Time_zone +#include "sql_load.h" // mysql_load +#include "sql_db.h" // load_db_opt_by_name +#include "slave.h" +#include "rpl_rli.h" +#include "rpl_mi.h" +#include "rpl_filter.h" +#include "rpl_record.h" +#include "transaction.h" +#include <my_dir.h> +#include "sql_show.h" // append_identifier +#include <mysql/psi/mysql_statement.h> +#include <strfunc.h> +#include "compat56.h" +#include "sql_insert.h" +#ifdef WITH_WSREP +#include "wsrep_mysqld.h" +#endif /* WITH_WSREP */ +#else +#include "mysqld_error.h" +#endif /* MYSQL_CLIENT */ + +#include <my_bitmap.h> +#include "rpl_utility.h" +#include "rpl_constants.h" +#include "sql_digest.h" +#include "zlib.h" +#include "myisampack.h" +#include <algorithm> + +#define my_b_write_string(A, B) my_b_write((A), (uchar*)(B), (uint) (sizeof(B) - 1)) + +#ifndef _AIX +PSI_memory_key key_memory_log_event; +#endif +PSI_memory_key key_memory_Incident_log_event_message; +PSI_memory_key key_memory_Rows_query_log_event_rows_query; + +/** + BINLOG_CHECKSUM variable. +*/ +const char *binlog_checksum_type_names[]= { + "NONE", + "CRC32", + NullS +}; + +unsigned int binlog_checksum_type_length[]= { + sizeof("NONE") - 1, + sizeof("CRC32") - 1, + 0 +}; + +TYPELIB binlog_checksum_typelib= +{ + array_elements(binlog_checksum_type_names) - 1, "", + binlog_checksum_type_names, + binlog_checksum_type_length +}; + + +#define FLAGSTR(V,F) ((V)&(F)?#F" ":"") + +/* + Size of buffer for printing a double in format %.<PREC>g + + optional '-' + optional zero + '.' + PREC digits + 'e' + sign + + exponent digits + '\0' +*/ +#define FMT_G_BUFSIZE(PREC) (3 + (PREC) + 5 + 1) + +/* + replication event checksum is introduced in the following "checksum-home" version. + The checksum-aware servers extract FD's version to decide whether the FD event + carries checksum info. + + TODO: correct the constant when it has been determined + (which main tree to push and when) +*/ +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); + +/* + Cache that will automatically be written to a dedicated file on + destruction. + + DESCRIPTION + + */ +class Write_on_release_cache +{ +public: + enum flag + { + FLUSH_F + }; + + typedef unsigned short flag_set; + + /* + Constructor. + + SYNOPSIS + Write_on_release_cache + cache Pointer to cache to use + file File to write cache to upon destruction + flags Flags for the cache + + DESCRIPTION + Cache common parameters and ensure common flush_data() code + on successful copy of the cache, the cache will be reinited as a + WRITE_CACHE. + + Currently, a pointer to the cache is provided in the + constructor, but it would be possible to create a subclass + holding the IO_CACHE itself. + */ + Write_on_release_cache(IO_CACHE *cache, FILE *file, flag_set flags= 0, Log_event *ev= NULL) + : m_cache(cache), m_file(file), m_flags(flags), m_ev(ev) + { + reinit_io_cache(m_cache, WRITE_CACHE, 0L, FALSE, TRUE); + } + + ~Write_on_release_cache() = default; + + bool flush_data() + { +#ifdef MYSQL_CLIENT + if (m_ev == NULL) + { + if (copy_event_cache_to_file_and_reinit(m_cache, m_file)) + return 1; + if ((m_flags & FLUSH_F) && fflush(m_file)) + return 1; + } + else // if m_ev<>NULL, then storing the output in output_buf + { + LEX_STRING tmp_str; + bool res; + if (copy_event_cache_to_string_and_reinit(m_cache, &tmp_str)) + return 1; + /* use 2 argument append as tmp_str is not \0 terminated */ + res= m_ev->output_buf.append(tmp_str.str, tmp_str.length); + my_free(tmp_str.str); + return res ? res : 0; + } +#else /* MySQL_SERVER */ + if (copy_event_cache_to_file_and_reinit(m_cache, m_file)) + return 1; + if ((m_flags & FLUSH_F) && fflush(m_file)) + return 1; +#endif + return 0; + } + + /* + Return a pointer to the internal IO_CACHE. + + SYNOPSIS + operator&() + + DESCRIPTION + + Function to return a pointer to the internal cache, so that the + object can be treated as a IO_CACHE and used with the my_b_* + IO_CACHE functions + + RETURN VALUE + A pointer to the internal IO_CACHE. + */ + IO_CACHE *operator&() + { + return m_cache; + } + +private: + // Hidden, to prevent usage. + Write_on_release_cache(Write_on_release_cache const&); + + IO_CACHE *m_cache; + FILE *m_file; + flag_set m_flags; + Log_event *m_ev; // Used for Flashback +}; + +#ifndef DBUG_OFF +#define DBUG_DUMP_EVENT_BUF(B,L) \ + do { \ + const uchar *_buf=(uchar*)(B); \ + size_t _len=(L); \ + if (_len >= LOG_EVENT_MINIMAL_HEADER_LEN) \ + { \ + DBUG_PRINT("data", ("header: timestamp:%u type:%u server_id:%u len:%u log_pos:%u flags:%u", \ + uint4korr(_buf), _buf[EVENT_TYPE_OFFSET], \ + uint4korr(_buf+SERVER_ID_OFFSET), \ + uint4korr(_buf+EVENT_LEN_OFFSET), \ + uint4korr(_buf+LOG_POS_OFFSET), \ + uint4korr(_buf+FLAGS_OFFSET))); \ + DBUG_DUMP("data", _buf+LOG_EVENT_MINIMAL_HEADER_LEN, \ + _len-LOG_EVENT_MINIMAL_HEADER_LEN); \ + } \ + else \ + DBUG_DUMP("data", _buf, _len); \ + } while(0) +#else +#define DBUG_DUMP_EVENT_BUF(B,L) do { } while(0) +#endif + +/* + read_str() +*/ + +static inline bool read_str(const uchar **buf, const uchar *buf_end, + const char **str, uint8 *len) +{ + if (*buf + ((uint) **buf) >= buf_end) + return 1; + *len= (uint8) **buf; + *str= (char*) (*buf)+1; + (*buf)+= (uint) *len+1; + return 0; +} + + +/** + Transforms a string into "" or its expression in X'HHHH' form. +*/ + +char *str_to_hex(char *to, const char *from, size_t len) +{ + if (len) + { + *to++= 'X'; + *to++= '\''; + to= octet2hex(to, from, len); + *to++= '\''; + *to= '\0'; + } + else + to= strmov(to, "\"\""); + return to; // pointer to end 0 of 'to' +} + +#define BINLOG_COMPRESSED_HEADER_LEN 1 +#define BINLOG_COMPRESSED_ORIGINAL_LENGTH_MAX_BYTES 4 +/** + Compressed Record + Record Header: 1 Byte + 7 Bit: Always 1, mean compressed; + 4-6 Bit: Compressed algorithm - Always 0, means zlib + It maybe support other compression algorithm in the future. + 0-3 Bit: Bytes of "Record Original Length" + Record Original Length: 1-4 Bytes + Compressed Buf: +*/ + +/** + Get the length of compress content. +*/ + +uint32 binlog_get_compress_len(uint32 len) +{ + /* 5 for the begin content, 1 reserved for a '\0'*/ + return ALIGN_SIZE((BINLOG_COMPRESSED_HEADER_LEN + BINLOG_COMPRESSED_ORIGINAL_LENGTH_MAX_BYTES) + + compressBound(len) + 1); +} + +/** + Compress buf from 'src' to 'dst'. + + Note: 1) Then the caller should guarantee the length of 'dst', which + can be got by binlog_get_uncompress_len, is enough to hold + the content uncompressed. + 2) The 'comlen' should stored the length of 'dst', and it will + be set as the size of compressed content after return. + + return zero if successful, others otherwise. +*/ +int binlog_buf_compress(const uchar *src, uchar *dst, uint32 len, uint32 *comlen) +{ + uchar lenlen; + if (len & 0xFF000000) + { + dst[1]= uchar(len >> 24); + dst[2]= uchar(len >> 16); + dst[3]= uchar(len >> 8); + dst[4]= uchar(len); + lenlen= 4; + } + else if (len & 0x00FF0000) + { + dst[1]= uchar(len >> 16); + dst[2]= uchar(len >> 8); + dst[3]= uchar(len); + lenlen= 3; + } + else if (len & 0x0000FF00) + { + dst[1]= uchar(len >> 8); + dst[2]= uchar(len); + lenlen= 2; + } + else + { + dst[1]= uchar(len); + lenlen= 1; + } + dst[0]= 0x80 | (lenlen & 0x07); + + uLongf tmplen= (uLongf)*comlen - BINLOG_COMPRESSED_HEADER_LEN - lenlen - 1; + if (compress((Bytef *)dst + BINLOG_COMPRESSED_HEADER_LEN + lenlen, &tmplen, + (const Bytef *)src, (uLongf)len) != Z_OK) + { + return 1; + } + *comlen= (uint32)tmplen + BINLOG_COMPRESSED_HEADER_LEN + lenlen; + return 0; +} + +/** + Convert a query_compressed_log_event to query_log_event + from 'src' to 'dst', the size after compression stored in 'newlen'. + + @Note: + 1) The caller should call my_free to release 'dst' if *is_malloc is + returned as true. + 2) If *is_malloc is retuened as false, then 'dst' reuses the passed-in + 'buf'. + + return zero if successful, non-zero otherwise. +*/ + +int +query_event_uncompress(const Format_description_log_event *description_event, + bool contain_checksum, const uchar *src, ulong src_len, + uchar* buf, ulong buf_size, bool* is_malloc, uchar **dst, + ulong *newlen) +{ + ulong len= uint4korr(src + EVENT_LEN_OFFSET); + const uchar *tmp= src; + const uchar *end= src + len; + uchar *new_dst; + + // bad event + if (src_len < len) + return 1; + + DBUG_ASSERT((uchar)src[EVENT_TYPE_OFFSET] == QUERY_COMPRESSED_EVENT); + + uint8 common_header_len= description_event->common_header_len; + uint8 post_header_len= + description_event->post_header_len[QUERY_COMPRESSED_EVENT-1]; + + *is_malloc= false; + + tmp+= common_header_len; + // bad event + if (end <= tmp) + return 1; + + uint db_len= (uint)tmp[Q_DB_LEN_OFFSET]; + uint16 status_vars_len= uint2korr(tmp + Q_STATUS_VARS_LEN_OFFSET); + + tmp+= post_header_len + status_vars_len + db_len + 1; + // bad event + if (end <= tmp) + return 1; + + int32 comp_len= (int32)(len - (tmp - src) - + (contain_checksum ? BINLOG_CHECKSUM_LEN : 0)); + uint32 un_len= binlog_get_uncompress_len(tmp); + + // bad event + if (comp_len < 0 || un_len == 0) + return 1; + + *newlen= (ulong)(tmp - src) + un_len; + if (contain_checksum) + *newlen+= BINLOG_CHECKSUM_LEN; + + uint32 alloc_size= (uint32)ALIGN_SIZE(*newlen); + + if (alloc_size <= buf_size) + new_dst= buf; + else + { + new_dst= (uchar *) my_malloc(PSI_INSTRUMENT_ME, alloc_size, MYF(MY_WME)); + if (!new_dst) + return 1; + *is_malloc= true; + } + + /* copy the head*/ + memcpy(new_dst, src , tmp - src); + if (binlog_buf_uncompress(tmp, new_dst + (tmp - src), comp_len, &un_len)) + { + if (*is_malloc) + { + *is_malloc= false; + my_free(new_dst); + } + return 1; + } + + new_dst[EVENT_TYPE_OFFSET]= QUERY_EVENT; + int4store(new_dst + EVENT_LEN_OFFSET, *newlen); + if (contain_checksum) + { + ulong clear_len= *newlen - BINLOG_CHECKSUM_LEN; + int4store(new_dst + clear_len, + my_checksum(0L, (uchar *)new_dst, clear_len)); + } + *dst= new_dst; + return 0; +} + +int +row_log_event_uncompress(const Format_description_log_event *description_event, + bool contain_checksum, const uchar *src, ulong src_len, + uchar* buf, ulong buf_size, bool* is_malloc, + uchar **dst, ulong *newlen) +{ + Log_event_type type= (Log_event_type)(uchar)src[EVENT_TYPE_OFFSET]; + ulong len= uint4korr(src + EVENT_LEN_OFFSET); + const uchar *tmp= src; + uchar *new_dst= NULL; + const uchar *end= tmp + len; + + if (src_len < len) + return 1; // bad event + + DBUG_ASSERT(LOG_EVENT_IS_ROW_COMPRESSED(type)); + + uint8 common_header_len= description_event->common_header_len; + uint8 post_header_len= description_event->post_header_len[type-1]; + + tmp+= common_header_len + ROWS_HEADER_LEN_V1; + if (post_header_len == ROWS_HEADER_LEN_V2) + { + /* + Have variable length header, check length, + which includes length bytes + */ + + if (end - tmp <= 2) + return 1; // bad event + + uint16 var_header_len= uint2korr(tmp); + DBUG_ASSERT(var_header_len >= 2); + + /* skip over var-len header, extracting 'chunks' */ + tmp+= var_header_len; + + /* get the uncompressed event type */ + type= + (Log_event_type)(type - WRITE_ROWS_COMPRESSED_EVENT + WRITE_ROWS_EVENT); + } + else + { + /* get the uncompressed event type */ + type= (Log_event_type) + (type - WRITE_ROWS_COMPRESSED_EVENT_V1 + WRITE_ROWS_EVENT_V1); + } + + if (end <= tmp) + return 1; //bad event + + ulong m_width= net_field_length((uchar **)&tmp); + tmp+= (m_width + 7) / 8; + + if (type == UPDATE_ROWS_EVENT_V1 || type == UPDATE_ROWS_EVENT) + { + tmp+= (m_width + 7) / 8; + } + + if (end <= tmp) + return 1; //bad event + + uint32 un_len= binlog_get_uncompress_len(tmp); + if (un_len == 0) + return 1; //bad event + + int32 comp_len= (int32)(len - (tmp - src) - + (contain_checksum ? BINLOG_CHECKSUM_LEN : 0)); + if (comp_len <=0) + return 1; //bad event + + *newlen= ulong(tmp - src) + un_len; + if (contain_checksum) + *newlen+= BINLOG_CHECKSUM_LEN; + + size_t alloc_size= ALIGN_SIZE(*newlen); + + *is_malloc= false; + if (alloc_size <= buf_size) + { + new_dst= buf; + } + else + { + new_dst= (uchar*) my_malloc(PSI_INSTRUMENT_ME, alloc_size, MYF(MY_WME)); + if (!new_dst) + return 1; + *is_malloc= true; + } + + /* Copy the head. */ + memcpy(new_dst, src , tmp - src); + /* Uncompress the body. */ + if (binlog_buf_uncompress(tmp, new_dst + (tmp - src), + comp_len, &un_len)) + { + if (*is_malloc) + my_free(new_dst); + return 1; + } + + new_dst[EVENT_TYPE_OFFSET]= type; + int4store(new_dst + EVENT_LEN_OFFSET, *newlen); + if (contain_checksum) + { + ulong clear_len= *newlen - BINLOG_CHECKSUM_LEN; + int4store(new_dst + clear_len, + my_checksum(0L, (uchar *)new_dst, clear_len)); + } + *dst= new_dst; + return 0; +} + +/** + Get the length of uncompress content. + return 0 means error. +*/ + +uint32 binlog_get_uncompress_len(const uchar *buf) +{ + uint32 len, lenlen; + + if ((buf == NULL) || ((buf[0] & 0xe0) != 0x80)) + return 0; + + lenlen= buf[0] & 0x07; + + buf++; + /* Length is stored in high byte first order, like myisam keys */ + switch(lenlen) { + case 1: + len= buf[0]; + break; + case 2: + len= mi_uint2korr(buf); + break; + case 3: + len= mi_uint3korr(buf); + break; + case 4: + len= mi_uint4korr(buf); + break; + default: + DBUG_ASSERT(lenlen >= 1 && lenlen <= 4); + len= 0; + break; + } + return len; +} + +/** + Uncompress the content in 'src' with length of 'len' to 'dst'. + + Note: 1) Then the caller should guarantee the length of 'dst' (which + can be got by statement_get_uncompress_len) is enough to hold + the content uncompressed. + 2) The 'newlen' should stored the length of 'dst', and it will + be set as the size of uncompressed content after return. + + return zero if successful, others otherwise. +*/ +int binlog_buf_uncompress(const uchar *src, uchar *dst, uint32 len, + uint32 *newlen) +{ + if ((src[0] & 0x80) == 0) + return 1; + + uint32 lenlen= src[0] & 0x07; + uLongf buflen= *newlen; // zlib type + + uint32 alg= (src[0] & 0x70) >> 4; + switch(alg) { + case 0: + // zlib + if (uncompress((Bytef *)dst, &buflen, + (const Bytef*)src + 1 + lenlen, len - 1 - lenlen) != Z_OK) + return 1; + break; + default: + //TODO + //bad algorithm + return 1; + } + + DBUG_ASSERT(*newlen == (uint32)buflen); + *newlen= (uint32)buflen; + return 0; +} + + +/************************************************************************** + Log_event methods (= the parent class of all events) +**************************************************************************/ + +/** + @return + returns the human readable name of the event's type +*/ + +const char* Log_event::get_type_str(Log_event_type type) +{ + switch(type) { + case START_EVENT_V3: return "Start_v3"; + case STOP_EVENT: return "Stop"; + case QUERY_EVENT: return "Query"; + case ROTATE_EVENT: return "Rotate"; + case INTVAR_EVENT: return "Intvar"; + case LOAD_EVENT: return "Load"; + case NEW_LOAD_EVENT: return "New_load"; + case SLAVE_EVENT: return "Slave"; + case CREATE_FILE_EVENT: return "Create_file"; + case APPEND_BLOCK_EVENT: return "Append_block"; + case DELETE_FILE_EVENT: return "Delete_file"; + case EXEC_LOAD_EVENT: return "Exec_load"; + case RAND_EVENT: return "RAND"; + case XID_EVENT: return "Xid"; + case USER_VAR_EVENT: return "User var"; + case FORMAT_DESCRIPTION_EVENT: return "Format_desc"; + case TABLE_MAP_EVENT: return "Table_map"; + case PRE_GA_WRITE_ROWS_EVENT: return "Write_rows_event_old"; + case PRE_GA_UPDATE_ROWS_EVENT: return "Update_rows_event_old"; + case PRE_GA_DELETE_ROWS_EVENT: return "Delete_rows_event_old"; + case WRITE_ROWS_EVENT_V1: return "Write_rows_v1"; + case UPDATE_ROWS_EVENT_V1: return "Update_rows_v1"; + case DELETE_ROWS_EVENT_V1: return "Delete_rows_v1"; + case WRITE_ROWS_EVENT: return "Write_rows"; + case UPDATE_ROWS_EVENT: return "Update_rows"; + case DELETE_ROWS_EVENT: return "Delete_rows"; + case BEGIN_LOAD_QUERY_EVENT: return "Begin_load_query"; + case EXECUTE_LOAD_QUERY_EVENT: return "Execute_load_query"; + case INCIDENT_EVENT: return "Incident"; + case ANNOTATE_ROWS_EVENT: return "Annotate_rows"; + case BINLOG_CHECKPOINT_EVENT: return "Binlog_checkpoint"; + case GTID_EVENT: return "Gtid"; + case GTID_LIST_EVENT: return "Gtid_list"; + case START_ENCRYPTION_EVENT: return "Start_encryption"; + + /* The following is only for mysqlbinlog */ + case IGNORABLE_LOG_EVENT: return "Ignorable log event"; + case ROWS_QUERY_LOG_EVENT: return "MySQL Rows_query"; + case GTID_LOG_EVENT: return "MySQL Gtid"; + case ANONYMOUS_GTID_LOG_EVENT: return "MySQL Anonymous_Gtid"; + case PREVIOUS_GTIDS_LOG_EVENT: return "MySQL Previous_gtids"; + case HEARTBEAT_LOG_EVENT: return "Heartbeat"; + case TRANSACTION_CONTEXT_EVENT: return "Transaction_context"; + case VIEW_CHANGE_EVENT: return "View_change"; + case XA_PREPARE_LOG_EVENT: return "XA_prepare"; + case QUERY_COMPRESSED_EVENT: return "Query_compressed"; + case WRITE_ROWS_COMPRESSED_EVENT: return "Write_rows_compressed"; + case UPDATE_ROWS_COMPRESSED_EVENT: return "Update_rows_compressed"; + case DELETE_ROWS_COMPRESSED_EVENT: return "Delete_rows_compressed"; + case WRITE_ROWS_COMPRESSED_EVENT_V1: return "Write_rows_compressed_v1"; + case UPDATE_ROWS_COMPRESSED_EVENT_V1: return "Update_rows_compressed_v1"; + case DELETE_ROWS_COMPRESSED_EVENT_V1: return "Delete_rows_compressed_v1"; + + default: return "Unknown"; /* impossible */ + } +} + +const char* Log_event::get_type_str() +{ + return get_type_str(get_type_code()); +} + + +/* + Log_event::Log_event() +*/ + +Log_event::Log_event(const uchar *buf, + const Format_description_log_event* description_event) + :temp_buf(0), exec_time(0), cache_type(Log_event::EVENT_INVALID_CACHE), + checksum_alg(BINLOG_CHECKSUM_ALG_UNDEF) +{ +#ifndef MYSQL_CLIENT + thd= 0; +#endif + when= uint4korr(buf); + when_sec_part= ~0UL; + server_id= uint4korr(buf + SERVER_ID_OFFSET); + data_written= uint4korr(buf + EVENT_LEN_OFFSET); + if (description_event->binlog_version==1) + { + log_pos= 0; + flags= 0; + return; + } + /* 4.0 or newer */ + log_pos= uint4korr(buf + LOG_POS_OFFSET); + /* + If the log is 4.0 (so here it can only be a 4.0 relay log read by + the SQL thread or a 4.0 master binlog read by the I/O thread), + log_pos is the beginning of the event: we transform it into the end + of the event, which is more useful. + But how do you know that the log is 4.0: you know it if + description_event is version 3 *and* you are not reading a + Format_desc (remember that mysqlbinlog starts by assuming that 5.0 + logs are in 4.0 format, until it finds a Format_desc). + */ + if (description_event->binlog_version==3 && + (uchar)buf[EVENT_TYPE_OFFSET]<FORMAT_DESCRIPTION_EVENT && log_pos) + { + /* + If log_pos=0, don't change it. log_pos==0 is a marker to mean + "don't change rli->group_master_log_pos" (see + inc_group_relay_log_pos()). As it is unreal log_pos, adding the + event len's is nonsense. For example, a fake Rotate event should + not have its log_pos (which is 0) changed or it will modify + Exec_master_log_pos in SHOW SLAVE STATUS, displaying a nonsense + value of (a non-zero offset which does not exist in the master's + binlog, so which will cause problems if the user uses this value + in CHANGE MASTER). + */ + log_pos+= data_written; /* purecov: inspected */ + } + DBUG_PRINT("info", ("log_pos: %llu", log_pos)); + + flags= uint2korr(buf + FLAGS_OFFSET); + if (((uchar)buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT) || + ((uchar)buf[EVENT_TYPE_OFFSET] == ROTATE_EVENT)) + { + /* + These events always have a header which stops here (i.e. their + header is FROZEN). + */ + /* + Initialization to zero of all other Log_event members as they're + not specified. Currently there are no such members; in the future + there will be an event UID (but Format_description and Rotate + don't need this UID, as they are not propagated through + --log-slave-updates (remember the UID is used to not play a query + twice when you have two masters which are slaves of a 3rd master). + Then we are done. + */ + return; + } + /* otherwise, go on with reading the header from buf (nothing now) */ +} + + +/** + This needn't be format-tolerant, because we only parse the first + LOG_EVENT_MINIMAL_HEADER_LEN bytes (just need the event's length). +*/ + +int Log_event::read_log_event(IO_CACHE* file, String* packet, + const Format_description_log_event *fdle, + enum enum_binlog_checksum_alg checksum_alg_arg) +{ + ulong data_len; + char buf[LOG_EVENT_MINIMAL_HEADER_LEN]; + uchar ev_offset= packet->length(); +#if !defined(MYSQL_CLIENT) + THD *thd=current_thd; + ulong max_allowed_packet= thd ? thd->slave_thread ? slave_max_allowed_packet + : thd->variables.max_allowed_packet + : ~(uint)0; +#endif + DBUG_ENTER("Log_event::read_log_event(IO_CACHE*,String*...)"); + + if (my_b_read(file, (uchar*) buf, sizeof(buf))) + { + /* + If the read hits eof, we must report it as eof so the caller + will know it can go into cond_wait to be woken up on the next + update to the log. + */ + DBUG_PRINT("error",("file->error: %d", file->error)); + DBUG_RETURN(file->error == 0 ? LOG_READ_EOF : + file->error > 0 ? LOG_READ_TRUNC : LOG_READ_IO); + } + data_len= uint4korr(buf + EVENT_LEN_OFFSET); + + /* Append the log event header to packet */ + if (packet->append(buf, sizeof(buf))) + DBUG_RETURN(LOG_READ_MEM); + + if (data_len < LOG_EVENT_MINIMAL_HEADER_LEN) + DBUG_RETURN(LOG_READ_BOGUS); + + if (data_len > MY_MAX(max_allowed_packet, + opt_binlog_rows_event_max_size + MAX_LOG_EVENT_HEADER)) + DBUG_RETURN(LOG_READ_TOO_LARGE); + + if (likely(data_len > LOG_EVENT_MINIMAL_HEADER_LEN)) + { + /* Append rest of event, read directly from file into packet */ + if (packet->append(file, data_len - LOG_EVENT_MINIMAL_HEADER_LEN)) + { + /* + Fatal error occurred when appending rest of the event + to packet, possible failures: + 1. EOF occurred when reading from file, it's really an error + as there's supposed to be more bytes available. + file->error will have been set to number of bytes left to read + 2. Read was interrupted, file->error would normally be set to -1 + 3. Failed to allocate memory for packet, my_errno + will be ENOMEM(file->error should be 0, but since the + memory allocation occurs before the call to read it might + be uninitialized) + */ + DBUG_RETURN(my_errno == ENOMEM ? LOG_READ_MEM : + (file->error >= 0 ? LOG_READ_TRUNC: LOG_READ_IO)); + } + } + + if (fdle->crypto_data.scheme) + { + uchar iv[BINLOG_IV_LENGTH]; + fdle->crypto_data.set_iv(iv, (uint32) (my_b_tell(file) - data_len)); + size_t sz= data_len + ev_offset + 1; +#ifdef HAVE_WOLFSSL + /* + Workaround for MDEV-19582. + WolfSSL reads memory out of bounds with decryption/NOPAD) + We allocate a little more memory therefore. + */ + sz+= MY_AES_BLOCK_SIZE; +#endif + char *newpkt= (char*)my_malloc(PSI_INSTRUMENT_ME, sz, MYF(MY_WME)); + if (!newpkt) + DBUG_RETURN(LOG_READ_MEM); + memcpy(newpkt, packet->ptr(), ev_offset); + + uint dstlen; + uchar *src= (uchar*)packet->ptr() + ev_offset; + uchar *dst= (uchar*)newpkt + ev_offset; + memcpy(src + EVENT_LEN_OFFSET, src, 4); + if (encryption_crypt(src + 4, data_len - 4, dst + 4, &dstlen, + fdle->crypto_data.key, fdle->crypto_data.key_length, iv, + sizeof(iv), ENCRYPTION_FLAG_DECRYPT | ENCRYPTION_FLAG_NOPAD, + ENCRYPTION_KEY_SYSTEM_DATA, fdle->crypto_data.key_version)) + { + my_free(newpkt); + DBUG_RETURN(LOG_READ_DECRYPT); + } + DBUG_ASSERT(dstlen == data_len - 4); + memcpy(dst, dst + EVENT_LEN_OFFSET, 4); + int4store(dst + EVENT_LEN_OFFSET, data_len); + packet->reset(newpkt, data_len + ev_offset, data_len + ev_offset + 1, + &my_charset_bin); + } + + /* + CRC verification of the Dump thread + */ + if (data_len > LOG_EVENT_MINIMAL_HEADER_LEN) + { + /* Corrupt the event for Dump thread*/ + DBUG_EXECUTE_IF("corrupt_read_log_event2", + uchar *debug_event_buf_c= (uchar*) packet->ptr() + ev_offset; + if (debug_event_buf_c[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT) + { + int debug_cor_pos= rand() % (data_len - BINLOG_CHECKSUM_LEN); + debug_event_buf_c[debug_cor_pos] =~ debug_event_buf_c[debug_cor_pos]; + DBUG_PRINT("info", ("Corrupt the event at Log_event::read_log_event: byte on position %d", debug_cor_pos)); + DBUG_SET("-d,corrupt_read_log_event2"); + } + ); + if (event_checksum_test((uchar*) packet->ptr() + ev_offset, + data_len, checksum_alg_arg)) + DBUG_RETURN(LOG_READ_CHECKSUM_FAILURE); + } + DBUG_RETURN(0); +} + +Log_event* Log_event::read_log_event(IO_CACHE* file, + const Format_description_log_event *fdle, + my_bool crc_check) +{ + DBUG_ENTER("Log_event::read_log_event(IO_CACHE*,Format_description_log_event*...)"); + DBUG_ASSERT(fdle != 0); + String event; + const char *error= 0; + Log_event *res= 0; + + switch (read_log_event(file, &event, fdle, BINLOG_CHECKSUM_ALG_OFF)) + { + case 0: + break; + case LOG_READ_EOF: // no error here; we are at the file's end + goto err; + case LOG_READ_BOGUS: + error= "Event invalid"; + goto err; + case LOG_READ_IO: + error= "read error"; + goto err; + case LOG_READ_MEM: + error= "Out of memory"; + goto err; + case LOG_READ_TRUNC: + error= "Event truncated"; + goto err; + case LOG_READ_TOO_LARGE: + error= "Event too big"; + goto err; + case LOG_READ_DECRYPT: + error= "Event decryption failure"; + goto err; + case LOG_READ_CHECKSUM_FAILURE: + default: + DBUG_ASSERT(0); + error= "internal error"; + goto err; + } + + if ((res= read_log_event((uchar*) event.ptr(), event.length(), + &error, fdle, crc_check))) + res->register_temp_buf((uchar*) event.release(), true); + +err: + if (unlikely(error)) + { + DBUG_ASSERT(!res); +#ifdef MYSQL_CLIENT + if (force_opt) + DBUG_RETURN(new Unknown_log_event()); +#endif + if (event.length() >= OLD_HEADER_LEN) + sql_print_error("Error in Log_event::read_log_event(): '%s'," + " data_len: %lu, event_type: %u", error, + (ulong) uint4korr(&event[EVENT_LEN_OFFSET]), + (uint) (uchar)event[EVENT_TYPE_OFFSET]); + else + sql_print_error("Error in Log_event::read_log_event(): '%s'", error); + /* + The SQL slave thread will check if file->error<0 to know + if there was an I/O error. Even if there is no "low-level" I/O errors + with 'file', any of the high-level above errors is worrying + enough to stop the SQL thread now ; as we are skipping the current event, + going on with reading and successfully executing other events can + only corrupt the slave's databases. So stop. + */ + file->error= -1; + } + DBUG_RETURN(res); +} + + +/** + Binlog format tolerance is in (buf, event_len, fdle) + constructors. +*/ + +Log_event* Log_event::read_log_event(const uchar *buf, uint event_len, + const char **error, + const Format_description_log_event *fdle, + my_bool crc_check) +{ + Log_event* ev; + enum enum_binlog_checksum_alg alg; + DBUG_ENTER("Log_event::read_log_event(char*,...)"); + DBUG_ASSERT(fdle != 0); + DBUG_PRINT("info", ("binlog_version: %d", fdle->binlog_version)); + DBUG_DUMP_EVENT_BUF(buf, event_len); + + /* + Check the integrity; This is needed because handle_slave_io() doesn't + check if packet is of proper length. + */ + if (event_len < EVENT_LEN_OFFSET) + { + *error="Sanity check failed"; // Needed to free buffer + DBUG_RETURN(NULL); // general sanity check - will fail on a partial read + } + + uint event_type= buf[EVENT_TYPE_OFFSET]; + // all following START events in the current file are without checksum + if (event_type == START_EVENT_V3) + (const_cast< Format_description_log_event *>(fdle))->checksum_alg= BINLOG_CHECKSUM_ALG_OFF; + /* + CRC verification by SQL and Show-Binlog-Events master side. + The caller has to provide @fdle->checksum_alg to + be the last seen FD's (A) descriptor. + If event is FD the descriptor is in it. + Notice, FD of the binlog can be only in one instance and therefore + Show-Binlog-Events executing master side thread needs just to know + the only FD's (A) value - whereas RL can contain more. + In the RL case, the alg is kept in FD_e (@fdle) which is reset + to the newer read-out event after its execution with possibly new alg descriptor. + Therefore in a typical sequence of RL: + {FD_s^0, FD_m, E_m^1} E_m^1 + will be verified with (A) of FD_m. + + See legends definition on MYSQL_BIN_LOG::relay_log_checksum_alg docs + lines (log.h). + + Notice, a pre-checksum FD version forces alg := BINLOG_CHECKSUM_ALG_UNDEF. + */ + alg= (event_type != FORMAT_DESCRIPTION_EVENT) ? + fdle->checksum_alg : get_checksum_alg(buf, event_len); + // Emulate the corruption during reading an event + DBUG_EXECUTE_IF("corrupt_read_log_event_char", + if (event_type != FORMAT_DESCRIPTION_EVENT) + { + uchar *debug_event_buf_c= const_cast<uchar*>(buf); + int debug_cor_pos= rand() % (event_len - BINLOG_CHECKSUM_LEN); + debug_event_buf_c[debug_cor_pos]=~ debug_event_buf_c[debug_cor_pos]; + DBUG_PRINT("info", ("Corrupt the event at Log_event::read_log_event(char*,...): byte on position %d", debug_cor_pos)); + DBUG_SET("-d,corrupt_read_log_event_char"); + } + ); + if (crc_check && event_checksum_test(const_cast<uchar*>(buf), event_len, alg)) + { +#ifdef MYSQL_CLIENT + *error= "Event crc check failed! Most likely there is event corruption."; + if (force_opt) + { + ev= new Unknown_log_event(buf, fdle); + DBUG_RETURN(ev); + } + else + DBUG_RETURN(NULL); +#else + *error= ER_THD_OR_DEFAULT(current_thd, ER_BINLOG_READ_EVENT_CHECKSUM_FAILURE); + sql_print_error("%s", *error); + DBUG_RETURN(NULL); +#endif + } + + if (event_type > fdle->number_of_event_types && + event_type != FORMAT_DESCRIPTION_EVENT) + { + /* + It is unsafe to use the fdle if its post_header_len + array does not include the event type. + */ + DBUG_PRINT("error", ("event type %d found, but the current " + "Format_description_log_event supports only %d event " + "types", event_type, + fdle->number_of_event_types)); + ev= NULL; + } + else + { + /* + In some previuos versions (see comment in + Format_description_log_event::Format_description_log_event(char*,...)), + event types were assigned different id numbers than in the + present version. In order to replicate from such versions to the + present version, we must map those event type id's to our event + type id's. The mapping is done with the event_type_permutation + array, which was set up when the Format_description_log_event + was read. + */ + if (fdle->event_type_permutation) + { + int new_event_type= fdle->event_type_permutation[event_type]; + DBUG_PRINT("info", ("converting event type %d to %d (%s)", + event_type, new_event_type, + get_type_str((Log_event_type)new_event_type))); + event_type= new_event_type; + } + + if (alg != BINLOG_CHECKSUM_ALG_UNDEF && + (event_type == FORMAT_DESCRIPTION_EVENT || + alg != BINLOG_CHECKSUM_ALG_OFF)) + event_len= event_len - BINLOG_CHECKSUM_LEN; + + /* + Create an object of Ignorable_log_event for unrecognized sub-class. + So that SLAVE SQL THREAD will only update the position and continue. + We should look for this flag first instead of judging by event_type + Any event can be Ignorable_log_event if it has this flag on. + look into @note of Ignorable_log_event + */ + if (uint2korr(buf + FLAGS_OFFSET) & LOG_EVENT_IGNORABLE_F) + { + ev= new Ignorable_log_event(buf, fdle, + get_type_str((Log_event_type) event_type)); + goto exit; + } + switch(event_type) { + case QUERY_EVENT: + ev= new Query_log_event(buf, event_len, fdle, QUERY_EVENT); + break; + case QUERY_COMPRESSED_EVENT: + ev= new Query_compressed_log_event(buf, event_len, fdle, + QUERY_COMPRESSED_EVENT); + break; + case LOAD_EVENT: + ev= new Load_log_event(buf, event_len, fdle); + break; + case NEW_LOAD_EVENT: + ev= new Load_log_event(buf, event_len, fdle); + break; + case ROTATE_EVENT: + ev= new Rotate_log_event(buf, event_len, fdle); + break; + case BINLOG_CHECKPOINT_EVENT: + ev= new Binlog_checkpoint_log_event(buf, event_len, fdle); + break; + case GTID_EVENT: + ev= new Gtid_log_event(buf, event_len, fdle); + break; + case GTID_LIST_EVENT: + ev= new Gtid_list_log_event(buf, event_len, fdle); + break; + case CREATE_FILE_EVENT: + ev= new Create_file_log_event(buf, event_len, fdle); + break; + case APPEND_BLOCK_EVENT: + ev= new Append_block_log_event(buf, event_len, fdle); + break; + case DELETE_FILE_EVENT: + ev= new Delete_file_log_event(buf, event_len, fdle); + break; + case EXEC_LOAD_EVENT: + ev= new Execute_load_log_event(buf, event_len, fdle); + break; + case START_EVENT_V3: /* this is sent only by MySQL <=4.x */ + ev= new Start_log_event_v3(buf, event_len, fdle); + break; + case STOP_EVENT: + ev= new Stop_log_event(buf, fdle); + break; + case INTVAR_EVENT: + ev= new Intvar_log_event(buf, fdle); + break; + case XID_EVENT: + ev= new Xid_log_event(buf, fdle); + break; + case XA_PREPARE_LOG_EVENT: + ev= new XA_prepare_log_event(buf, fdle); + break; + case RAND_EVENT: + ev= new Rand_log_event(buf, fdle); + break; + case USER_VAR_EVENT: + ev= new User_var_log_event(buf, event_len, fdle); + break; + case FORMAT_DESCRIPTION_EVENT: + ev= new Format_description_log_event(buf, event_len, fdle); + break; +#if defined(HAVE_REPLICATION) + case PRE_GA_WRITE_ROWS_EVENT: + ev= new Write_rows_log_event_old(buf, event_len, fdle); + break; + case PRE_GA_UPDATE_ROWS_EVENT: + ev= new Update_rows_log_event_old(buf, event_len, fdle); + break; + case PRE_GA_DELETE_ROWS_EVENT: + ev= new Delete_rows_log_event_old(buf, event_len, fdle); + break; + case WRITE_ROWS_EVENT_V1: + case WRITE_ROWS_EVENT: + ev= new Write_rows_log_event(buf, event_len, fdle); + break; + case UPDATE_ROWS_EVENT_V1: + case UPDATE_ROWS_EVENT: + ev= new Update_rows_log_event(buf, event_len, fdle); + break; + case DELETE_ROWS_EVENT_V1: + case DELETE_ROWS_EVENT: + ev= new Delete_rows_log_event(buf, event_len, fdle); + break; + + case WRITE_ROWS_COMPRESSED_EVENT: + case WRITE_ROWS_COMPRESSED_EVENT_V1: + ev= new Write_rows_compressed_log_event(buf, event_len, fdle); + break; + case UPDATE_ROWS_COMPRESSED_EVENT: + case UPDATE_ROWS_COMPRESSED_EVENT_V1: + ev= new Update_rows_compressed_log_event(buf, event_len, fdle); + break; + case DELETE_ROWS_COMPRESSED_EVENT: + case DELETE_ROWS_COMPRESSED_EVENT_V1: + ev= new Delete_rows_compressed_log_event(buf, event_len, fdle); + break; + + /* MySQL GTID events are ignored */ + case GTID_LOG_EVENT: + case ANONYMOUS_GTID_LOG_EVENT: + case PREVIOUS_GTIDS_LOG_EVENT: + case TRANSACTION_CONTEXT_EVENT: + case VIEW_CHANGE_EVENT: + ev= new Ignorable_log_event(buf, fdle, + get_type_str((Log_event_type) event_type)); + break; + + case TABLE_MAP_EVENT: + ev= new Table_map_log_event(buf, event_len, fdle); + break; +#endif + case BEGIN_LOAD_QUERY_EVENT: + ev= new Begin_load_query_log_event(buf, event_len, fdle); + break; + case EXECUTE_LOAD_QUERY_EVENT: + ev= new Execute_load_query_log_event(buf, event_len, fdle); + break; + case INCIDENT_EVENT: + ev= new Incident_log_event(buf, event_len, fdle); + break; + case ANNOTATE_ROWS_EVENT: + ev= new Annotate_rows_log_event(buf, event_len, fdle); + break; + case START_ENCRYPTION_EVENT: + ev= new Start_encryption_log_event(buf, event_len, fdle); + break; + default: + DBUG_PRINT("error",("Unknown event code: %d", + (uchar) buf[EVENT_TYPE_OFFSET])); + ev= NULL; + break; + } + } +exit: + + if (ev) + { + ev->checksum_alg= alg; +#ifdef MYSQL_CLIENT + if (ev->checksum_alg != BINLOG_CHECKSUM_ALG_OFF && + ev->checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF) + ev->crc= uint4korr(buf + (event_len)); +#endif + } + + DBUG_PRINT("read_event", ("%s(type_code: %u; event_len: %u)", + ev ? ev->get_type_str() : "<unknown>", + (uchar)buf[EVENT_TYPE_OFFSET], + event_len)); + /* + is_valid() are small event-specific sanity tests which are + important; for example there are some my_malloc() in constructors + (e.g. Query_log_event::Query_log_event(char*...)); when these + my_malloc() fail we can't return an error out of the constructor + (because constructor is "void") ; so instead we leave the pointer we + wanted to allocate (e.g. 'query') to 0 and we test it in is_valid(). + Same for Format_description_log_event, member 'post_header_len'. + + SLAVE_EVENT is never used, so it should not be read ever. + */ + if (!ev || !ev->is_valid() || (event_type == SLAVE_EVENT)) + { + DBUG_PRINT("error",("Found invalid event in binary log")); + + delete ev; +#ifdef MYSQL_CLIENT + if (!force_opt) /* then mysqlbinlog dies */ + { + *error= "Found invalid event in binary log"; + DBUG_RETURN(0); + } + ev= new Unknown_log_event(buf, fdle); +#else + *error= "Found invalid event in binary log"; + DBUG_RETURN(0); +#endif + } + DBUG_RETURN(ev); +} + + + +/* 2 utility functions for the next method */ + +/** + Read a string with length from memory. + + This function reads the string-with-length stored at + <code>src</code> and extract the length into <code>*len</code> and + a pointer to the start of the string into <code>*dst</code>. The + string can then be copied using <code>memcpy()</code> with the + number of bytes given in <code>*len</code>. + + @param src Pointer to variable holding a pointer to the memory to + read the string from. + @param dst Pointer to variable holding a pointer where the actual + string starts. Starting from this position, the string + can be copied using @c memcpy(). + @param len Pointer to variable where the length will be stored. + @param end One-past-the-end of the memory where the string is + stored. + + @return Zero if the entire string can be copied successfully, + @c UINT_MAX if the length could not be read from memory + (that is, if <code>*src >= end</code>), otherwise the + number of bytes that are missing to read the full + string, which happends <code>*dst + *len >= end</code>. +*/ +static int +get_str_len_and_pointer(const Log_event::Byte **src, + const char **dst, + uint *len, + const Log_event::Byte *end) +{ + if (*src >= end) + return -1; // Will be UINT_MAX in two-complement arithmetics + uint length= **src; + if (length > 0) + { + if (*src + length >= end) + return (int)(*src + length - end + 1); // Number of bytes missing + *dst= (char *)*src + 1; // Will be copied later + } + *len= length; + *src+= length + 1; + return 0; +} + +static void copy_str_and_move(const char **src, Log_event::Byte **dst, + size_t len) +{ + memcpy(*dst, *src, len); + *src= (const char *)*dst; + (*dst)+= len; + *(*dst)++= 0; +} + + +#ifdef DBUG_TRACE +static char const * +code_name(int code) +{ + static char buf[255]; + switch (code) { + case Q_FLAGS2_CODE: return "Q_FLAGS2_CODE"; + case Q_SQL_MODE_CODE: return "Q_SQL_MODE_CODE"; + case Q_CATALOG_CODE: return "Q_CATALOG_CODE"; + case Q_AUTO_INCREMENT: return "Q_AUTO_INCREMENT"; + case Q_CHARSET_CODE: return "Q_CHARSET_CODE"; + case Q_TIME_ZONE_CODE: return "Q_TIME_ZONE_CODE"; + case Q_CATALOG_NZ_CODE: return "Q_CATALOG_NZ_CODE"; + case Q_LC_TIME_NAMES_CODE: return "Q_LC_TIME_NAMES_CODE"; + case Q_CHARSET_DATABASE_CODE: return "Q_CHARSET_DATABASE_CODE"; + case Q_TABLE_MAP_FOR_UPDATE_CODE: return "Q_TABLE_MAP_FOR_UPDATE_CODE"; + case Q_MASTER_DATA_WRITTEN_CODE: return "Q_MASTER_DATA_WRITTEN_CODE"; + case Q_HRNOW: return "Q_HRNOW"; + case Q_XID: return "XID"; + case Q_GTID_FLAGS3: return "Q_GTID_FLAGS3"; + } + sprintf(buf, "CODE#%d", code); + return buf; +} +#endif + +#define VALIDATE_BYTES_READ(CUR_POS, START, EVENT_LEN) \ + do { \ + uchar *cur_pos= (uchar *)CUR_POS; \ + uchar *start= (uchar *)START; \ + uint len= EVENT_LEN; \ + uint bytes_read= (uint)(cur_pos - start); \ + DBUG_PRINT("info", ("Bytes read: %u event_len:%u.\n",\ + bytes_read, len)); \ + if (bytes_read >= len) \ + DBUG_VOID_RETURN; \ + } while (0) + +/** + Macro to check that there is enough space to read from memory. + + @param PTR Pointer to memory + @param END End of memory + @param CNT Number of bytes that should be read. + */ +#define CHECK_SPACE(PTR,END,CNT) \ + do { \ + DBUG_PRINT("info", ("Read %s", code_name(pos[-1]))); \ + if ((PTR) + (CNT) > (END)) { \ + DBUG_PRINT("info", ("query= 0")); \ + query= 0; \ + DBUG_VOID_RETURN; \ + } \ + } while (0) + + +/** + This is used by the SQL slave thread to prepare the event before execution. +*/ +Query_log_event::Query_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event, + Log_event_type event_type) + :Log_event(buf, description_event), data_buf(0), query(NullS), + db(NullS), catalog_len(0), status_vars_len(0), + flags2_inited(0), sql_mode_inited(0), charset_inited(0), flags2(0), + 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), xid(0), master_data_written(0), gtid_flags_extra(0), + sa_seq_no(0) +{ + ulong data_len; + uint32 tmp; + uint8 common_header_len, post_header_len; + Log_event::Byte *start; + const Log_event::Byte *end; + bool catalog_nz= 1; + DBUG_ENTER("Query_log_event::Query_log_event(char*,...)"); + + memset(&user, 0, sizeof(user)); + memset(&host, 0, sizeof(host)); + common_header_len= description_event->common_header_len; + post_header_len= description_event->post_header_len[event_type-1]; + DBUG_PRINT("info",("event_len: %u common_header_len: %d post_header_len: %d", + event_len, common_header_len, post_header_len)); + + /* + We test if the event's length is sensible, and if so we compute data_len. + We cannot rely on QUERY_HEADER_LEN here as it would not be format-tolerant. + We use QUERY_HEADER_MINIMAL_LEN which is the same for 3.23, 4.0 & 5.0. + */ + if (event_len < (uint)(common_header_len + post_header_len)) + DBUG_VOID_RETURN; + data_len= event_len - (common_header_len + post_header_len); + buf+= common_header_len; + + thread_id = slave_proxy_id = uint4korr(buf + Q_THREAD_ID_OFFSET); + exec_time = uint4korr(buf + Q_EXEC_TIME_OFFSET); + db_len = (uchar)buf[Q_DB_LEN_OFFSET]; // TODO: add a check of all *_len vars + error_code = uint2korr(buf + Q_ERR_CODE_OFFSET); + + /* + 5.0 format starts here. + Depending on the format, we may or not have affected/warnings etc + The remnent post-header to be parsed has length: + */ + tmp= post_header_len - QUERY_HEADER_MINIMAL_LEN; + if (tmp) + { + status_vars_len= uint2korr(buf + Q_STATUS_VARS_LEN_OFFSET); + /* + Check if status variable length is corrupt and will lead to very + wrong data. We could be even more strict and require data_len to + be even bigger, but this will suffice to catch most corruption + errors that can lead to a crash. + */ + if (status_vars_len > MY_MIN(data_len, MAX_SIZE_LOG_EVENT_STATUS)) + { + DBUG_PRINT("info", ("status_vars_len (%u) > data_len (%lu); query= 0", + status_vars_len, data_len)); + query= 0; + DBUG_VOID_RETURN; + } + data_len-= status_vars_len; + DBUG_PRINT("info", ("Query_log_event has status_vars_len: %u", + (uint) status_vars_len)); + tmp-= 2; + } + else + { + /* + server version < 5.0 / binlog_version < 4 master's event is + relay-logged with storing the original size of the event in + Q_MASTER_DATA_WRITTEN_CODE status variable. + The size is to be restored at reading Q_MASTER_DATA_WRITTEN_CODE-marked + event from the relay log. + */ + DBUG_ASSERT(description_event->binlog_version < 4); + master_data_written= (uint32)data_written; + } + /* + We have parsed everything we know in the post header for QUERY_EVENT, + the rest of post header is either comes from older version MySQL or + dedicated to derived events (e.g. Execute_load_query...) + */ + + /* variable-part: the status vars; only in MySQL 5.0 */ + + start= (Log_event::Byte*) (buf+post_header_len); + end= (const Log_event::Byte*) (start+status_vars_len); + for (const Log_event::Byte* pos= start; pos < end;) + { + switch (*pos++) { + case Q_FLAGS2_CODE: + CHECK_SPACE(pos, end, 4); + flags2_inited= description_event->options_written_to_bin_log; + flags2= uint4korr(pos); + DBUG_PRINT("info",("In Query_log_event, read flags2: %lu", (ulong) flags2)); + pos+= 4; + break; + case Q_SQL_MODE_CODE: + { + CHECK_SPACE(pos, end, 8); + sql_mode_inited= 1; + sql_mode= (sql_mode_t) uint8korr(pos); + DBUG_PRINT("info",("In Query_log_event, read sql_mode: %llu", sql_mode)); + pos+= 8; + break; + } + case Q_CATALOG_NZ_CODE: + DBUG_PRINT("info", ("case Q_CATALOG_NZ_CODE; pos:%p; end:%p", + pos, end)); + if (get_str_len_and_pointer(&pos, &catalog, &catalog_len, end)) + { + DBUG_PRINT("info", ("query= 0")); + query= 0; + DBUG_VOID_RETURN; + } + break; + case Q_AUTO_INCREMENT: + CHECK_SPACE(pos, end, 4); + auto_increment_increment= uint2korr(pos); + auto_increment_offset= uint2korr(pos+2); + pos+= 4; + break; + case Q_CHARSET_CODE: + { + CHECK_SPACE(pos, end, 6); + charset_inited= 1; + memcpy(charset, pos, 6); + pos+= 6; + break; + } + case Q_TIME_ZONE_CODE: + { + if (get_str_len_and_pointer(&pos, &time_zone_str, &time_zone_len, end)) + { + DBUG_PRINT("info", ("Q_TIME_ZONE_CODE: query= 0")); + query= 0; + DBUG_VOID_RETURN; + } + break; + } + case Q_CATALOG_CODE: /* for 5.0.x where 0<=x<=3 masters */ + CHECK_SPACE(pos, end, 1); + if ((catalog_len= *pos)) + catalog= (char*) pos+1; // Will be copied later + CHECK_SPACE(pos, end, catalog_len + 2); + pos+= catalog_len+2; // leap over end 0 + catalog_nz= 0; // catalog has end 0 in event + break; + case Q_LC_TIME_NAMES_CODE: + CHECK_SPACE(pos, end, 2); + lc_time_names_number= uint2korr(pos); + pos+= 2; + break; + case Q_CHARSET_DATABASE_CODE: + CHECK_SPACE(pos, end, 2); + charset_database_number= uint2korr(pos); + pos+= 2; + break; + case Q_TABLE_MAP_FOR_UPDATE_CODE: + CHECK_SPACE(pos, end, 8); + table_map_for_update= uint8korr(pos); + pos+= 8; + break; + case Q_MASTER_DATA_WRITTEN_CODE: + CHECK_SPACE(pos, end, 4); + data_written= master_data_written= uint4korr(pos); + pos+= 4; + break; + case Q_INVOKER: + { + CHECK_SPACE(pos, end, 1); + user.length= *pos++; + CHECK_SPACE(pos, end, user.length); + user.str= (char *)pos; + pos+= user.length; + + CHECK_SPACE(pos, end, 1); + host.length= *pos++; + CHECK_SPACE(pos, end, host.length); + host.str= (char *)pos; + pos+= host.length; + break; + } + case Q_HRNOW: + { + CHECK_SPACE(pos, end, 3); + when_sec_part= uint3korr(pos); + pos+= 3; + break; + } + case Q_XID: + { + CHECK_SPACE(pos, end, 8); + xid= uint8korr(pos); + pos+= 8; + break; + } + case Q_GTID_FLAGS3: + { + CHECK_SPACE(pos, end, 1); + gtid_flags_extra= *pos++; + if (gtid_flags_extra & (Gtid_log_event::FL_COMMIT_ALTER_E1 | + Gtid_log_event::FL_ROLLBACK_ALTER_E1)) + { + CHECK_SPACE(pos, end, 8); + sa_seq_no = uint8korr(pos); + pos+= 8; + } + break; + } + default: + /* That's why you must write status vars in growing order of code */ + DBUG_PRINT("info",("Query_log_event has unknown status vars (first has\ + code: %u), skipping the rest of them", (uint) *(pos-1))); + pos= (const uchar*) end; // Break loop + } + } + +#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 + +--------+-----------+------+------+---------+----+-------+ + | catlog | time_zone | user | host | db name | \0 | Query | + +--------+-----------+------+------+---------+----+-------+ + + To support the query cache we append the following buffer to the above + +-------+----------------------------------------+-------+ + |db len | uninitiatlized space of size of db len | FLAGS | + +-------+----------------------------------------+-------+ + + The area of buffer starting from Query field all the way to the end belongs + to the Query buffer and its structure is described in alloc_query() in + sql_parse.cc + */ + +#if !defined(MYSQL_CLIENT) && defined(HAVE_QUERY_CACHE) + if (!(start= data_buf= (Log_event::Byte*) my_malloc(PSI_INSTRUMENT_ME, + catalog_len + 1 + + time_zone_len + 1 + + user.length + 1 + + host.length + 1 + + data_len + 1 + + sizeof(size_t)//for db_len + + db_len + 1 + + QUERY_CACHE_DB_LENGTH_SIZE + + QUERY_CACHE_FLAGS_SIZE, + MYF(MY_WME)))) +#else + if (!(start= data_buf= (Log_event::Byte*) my_malloc(PSI_INSTRUMENT_ME, + catalog_len + 1 + + time_zone_len + 1 + + user.length + 1 + + host.length + 1 + + data_len + 1, + MYF(MY_WME)))) +#endif + DBUG_VOID_RETURN; + if (catalog_len) // If catalog is given + { + /** + @todo we should clean up and do only copy_str_and_move; it + works for both cases. Then we can remove the catalog_nz + flag. /sven + */ + if (likely(catalog_nz)) // true except if event comes from 5.0.0|1|2|3. + copy_str_and_move(&catalog, &start, catalog_len); + else + { + memcpy(start, catalog, catalog_len+1); // copy end 0 + catalog= (const char *)start; + start+= catalog_len+1; + } + } + if (time_zone_len) + copy_str_and_move(&time_zone_str, &start, time_zone_len); + + if (user.length) + { + copy_str_and_move(&user.str, &start, user.length); + } + else + { + user.str= (char*) start; + *(start++)= 0; + } + + if (host.length) + copy_str_and_move(&host.str, &start, host.length); + else + { + host.str= (char*) start; + *(start++)= 0; + } + + /** + if time_zone_len or catalog_len are 0, then time_zone and catalog + are uninitialized at this point. shouldn't they point to the + zero-length null-terminated strings we allocated space for in the + my_alloc call above? /sven + */ + + /* A 2nd variable part; this is common to all versions */ + memcpy((char*) start, end, data_len); // Copy db and query + start[data_len]= '\0'; // End query with \0 (For safetly) + db= (char *)start; + query= (char *)(start + db_len + 1); + q_len= data_len - db_len -1; + + if (data_len && (data_len < db_len || + data_len < q_len || + data_len != (db_len + q_len + 1))) + { + q_len= 0; + query= NULL; + DBUG_VOID_RETURN; + } + + uint32 max_length= uint32(event_len - ((end + db_len + 1) - + (buf - common_header_len))); + if (q_len != max_length || + (event_len < uint((end + db_len + 1) - (buf - common_header_len)))) + { + q_len= 0; + query= NULL; + DBUG_VOID_RETURN; + } + /** + Append the db length at the end of the buffer. This will be used by + Query_cache::send_result_to_client() in case the query cache is On. + */ +#if !defined(MYSQL_CLIENT) && defined(HAVE_QUERY_CACHE) + size_t db_length= (size_t)db_len; + memcpy(start + data_len + 1, &db_length, sizeof(size_t)); +#endif + DBUG_VOID_RETURN; +} + +Query_compressed_log_event::Query_compressed_log_event(const uchar *buf, + uint event_len, + const Format_description_log_event + *description_event, + Log_event_type event_type) + :Query_log_event(buf, event_len, description_event, event_type), + query_buf(NULL) +{ + if (query) + { + uint32 un_len= binlog_get_uncompress_len((uchar*) query); + if (!un_len) + { + query= 0; + return; + } + + /* Reserve one byte for '\0' */ + query_buf= (Log_event::Byte*) my_malloc(PSI_INSTRUMENT_ME, + ALIGN_SIZE(un_len + 1), MYF(MY_WME)); + if (query_buf && !binlog_buf_uncompress((uchar*) query, (uchar *) query_buf, + q_len, &un_len)) + { + query_buf[un_len]= 0; + query= (char*) query_buf; + q_len= un_len; + } + else + { + query= 0; + } + } +} + + +/* + Replace a binlog event read into a packet with a dummy event. Either a + Query_log_event that has just a comment, or if that will not fit in the + space used for the event to be replaced, then a NULL user_var event. + + This is used when sending binlog data to a slave which does not understand + this particular event and which is too old to support informational events + or holes in the event stream. + + This allows to write such events into the binlog on the master and still be + able to replicate against old slaves without them breaking. + + Clears the flag LOG_EVENT_THREAD_SPECIFIC_F and set LOG_EVENT_SUPPRESS_USE_F. + Overwrites the type with QUERY_EVENT (or USER_VAR_EVENT), and replaces the + body with a minimal query / NULL user var. + + Returns zero on success, -1 if error due to too little space in original + event. A minimum of 25 bytes (19 bytes fixed header + 6 bytes in the body) + is needed in any event to be replaced with a dummy event. +*/ +int +Query_log_event::dummy_event(String *packet, ulong ev_offset, + enum enum_binlog_checksum_alg checksum_alg) +{ + uchar *p= (uchar *)packet->ptr() + ev_offset; + size_t data_len= packet->length() - ev_offset; + uint16 flags; + static const size_t min_user_var_event_len= + LOG_EVENT_HEADER_LEN + UV_NAME_LEN_SIZE + 1 + UV_VAL_IS_NULL; // 25 + static const size_t min_query_event_len= + LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN + 1 + 1; // 34 + + if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32) + data_len-= BINLOG_CHECKSUM_LEN; + else + DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF || + checksum_alg == BINLOG_CHECKSUM_ALG_OFF); + + if (data_len < min_user_var_event_len) + /* Cannot replace with dummy, event too short. */ + return -1; + + flags= uint2korr(p + FLAGS_OFFSET); + flags&= ~LOG_EVENT_THREAD_SPECIFIC_F; + flags|= LOG_EVENT_SUPPRESS_USE_F; + int2store(p + FLAGS_OFFSET, flags); + + if (data_len < min_query_event_len) + { + /* + Have to use dummy user_var event for such a short packet. + + This works, but the event will be considered part of an event group with + the following event. So for example @@global.sql_slave_skip_counter=1 + will skip not only the dummy event, but also the immediately following + event. + + We write a NULL user var with the name @`!dummyvar` (or as much + as that as will fit within the size of the original event - so + possibly just @`!`). + */ + static const char var_name[]= "!dummyvar"; + size_t name_len= data_len - (min_user_var_event_len - 1); + + p[EVENT_TYPE_OFFSET]= USER_VAR_EVENT; + int4store(p + LOG_EVENT_HEADER_LEN, name_len); + memcpy(p + LOG_EVENT_HEADER_LEN + UV_NAME_LEN_SIZE, var_name, name_len); + p[LOG_EVENT_HEADER_LEN + UV_NAME_LEN_SIZE + name_len]= 1; // indicates NULL + } + else + { + /* + Use a dummy query event, just a comment. + */ + static const char message[]= + "# Dummy event replacing event type %u that slave cannot handle."; + char buf[sizeof(message)+1]; /* +1, as %u can expand to 3 digits. */ + uchar old_type= p[EVENT_TYPE_OFFSET]; + uchar *q= p + LOG_EVENT_HEADER_LEN; + size_t comment_len, len; + + p[EVENT_TYPE_OFFSET]= QUERY_EVENT; + int4store(q + Q_THREAD_ID_OFFSET, 0); + int4store(q + Q_EXEC_TIME_OFFSET, 0); + q[Q_DB_LEN_OFFSET]= 0; + int2store(q + Q_ERR_CODE_OFFSET, 0); + int2store(q + Q_STATUS_VARS_LEN_OFFSET, 0); + q[Q_DATA_OFFSET]= 0; /* Zero terminator for empty db */ + q+= Q_DATA_OFFSET + 1; + len= my_snprintf(buf, sizeof(buf), message, old_type); + comment_len= data_len - (min_query_event_len - 1); + if (comment_len <= len) + memcpy(q, buf, comment_len); + else + { + memcpy(q, buf, len); + memset(q+len, ' ', comment_len - len); + } + } + + if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32) + { + ha_checksum crc= my_checksum(0, p, data_len); + int4store(p + data_len, crc); + } + return 0; +} + +/* + Replace an event (GTID event) with a BEGIN query event, to be compatible + with an old slave. +*/ +int +Query_log_event::begin_event(String *packet, ulong ev_offset, + enum enum_binlog_checksum_alg checksum_alg) +{ + uchar *p= (uchar *)packet->ptr() + ev_offset; + uchar *q= p + LOG_EVENT_HEADER_LEN; + size_t data_len= packet->length() - ev_offset; + uint16 flags; + + if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32) + data_len-= BINLOG_CHECKSUM_LEN; + else + DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF || + checksum_alg == BINLOG_CHECKSUM_ALG_OFF); + + /* + Currently we only need to replace GTID event. + The length of GTID differs depending on whether it contains commit id. + */ + DBUG_ASSERT(data_len == LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN || + data_len == LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN + 2); + if (data_len != LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN && + data_len != LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN + 2) + return 1; + + flags= uint2korr(p + FLAGS_OFFSET); + flags&= ~LOG_EVENT_THREAD_SPECIFIC_F; + flags|= LOG_EVENT_SUPPRESS_USE_F; + int2store(p + FLAGS_OFFSET, flags); + + p[EVENT_TYPE_OFFSET]= QUERY_EVENT; + int4store(q + Q_THREAD_ID_OFFSET, 0); + int4store(q + Q_EXEC_TIME_OFFSET, 0); + q[Q_DB_LEN_OFFSET]= 0; + int2store(q + Q_ERR_CODE_OFFSET, 0); + if (data_len == LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN) + { + int2store(q + Q_STATUS_VARS_LEN_OFFSET, 0); + q[Q_DATA_OFFSET]= 0; /* Zero terminator for empty db */ + q+= Q_DATA_OFFSET + 1; + } + else + { + DBUG_ASSERT(data_len == LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN + 2); + /* Put in an empty time_zone_str to take up the extra 2 bytes. */ + int2store(q + Q_STATUS_VARS_LEN_OFFSET, 2); + q[Q_DATA_OFFSET]= Q_TIME_ZONE_CODE; + q[Q_DATA_OFFSET+1]= 0; /* Zero length for empty time_zone_str */ + q[Q_DATA_OFFSET+2]= 0; /* Zero terminator for empty db */ + q+= Q_DATA_OFFSET + 3; + } + memcpy(q, "BEGIN", 5); + + if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32) + { + ha_checksum crc= my_checksum(0, p, data_len); + int4store(p + data_len, crc); + } + return 0; +} + + +/************************************************************************** + Start_log_event_v3 methods +**************************************************************************/ + + +Start_log_event_v3::Start_log_event_v3(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + :Log_event(buf, description_event), binlog_version(BINLOG_VERSION) +{ + if (event_len < LOG_EVENT_MINIMAL_HEADER_LEN + ST_COMMON_HEADER_LEN_OFFSET) + { + server_version[0]= 0; + return; + } + buf+= LOG_EVENT_MINIMAL_HEADER_LEN; + binlog_version= uint2korr(buf+ST_BINLOG_VER_OFFSET); + memcpy(server_version, buf+ST_SERVER_VER_OFFSET, + ST_SERVER_VER_LEN); + // prevent overrun if log is corrupted on disk + server_version[ST_SERVER_VER_LEN-1]= 0; + created= uint4korr(buf+ST_CREATED_OFFSET); + dont_set_created= 1; +} + + +/*************************************************************************** + Format_description_log_event methods +****************************************************************************/ + +/** + Format_description_log_event 1st ctor. + + Ctor. Can be used to create the event to write to the binary log (when the + server starts or when FLUSH LOGS), or to create artificial events to parse + binlogs from MySQL 3.23 or 4.x. + When in a client, only the 2nd use is possible. + + @param binlog_version the binlog version for which we want to build + an event. Can be 1 (=MySQL 3.23), 3 (=4.0.x + x>=2 and 4.1) or 4 (MySQL 5.0). Note that the + old 4.0 (binlog version 2) is not supported; + it should not be used for replication with + 5.0. + @param server_ver a string containing the server version. +*/ + +Format_description_log_event:: +Format_description_log_event(uint8 binlog_ver, const char* server_ver) + :Start_log_event_v3(), event_type_permutation(0) +{ + binlog_version= binlog_ver; + switch (binlog_ver) { + case 4: /* MySQL 5.0 */ + memcpy(server_version, ::server_version, ST_SERVER_VER_LEN); + DBUG_EXECUTE_IF("pretend_version_50034_in_binlog", + strmov(server_version, "5.0.34");); + common_header_len= LOG_EVENT_HEADER_LEN; + number_of_event_types= LOG_EVENT_TYPES; + /* we'll catch my_malloc() error in is_valid() */ + post_header_len=(uint8*) my_malloc(PSI_INSTRUMENT_ME, + number_of_event_types*sizeof(uint8) + + BINLOG_CHECKSUM_ALG_DESC_LEN, + MYF(0)); + /* + This long list of assignments is not beautiful, but I see no way to + make it nicer, as the right members are #defines, not array members, so + it's impossible to write a loop. + */ + if (post_header_len) + { +#ifndef DBUG_OFF + // Allows us to sanity-check that all events initialized their + // events (see the end of this 'if' block). + memset(post_header_len, 255, number_of_event_types*sizeof(uint8)); +#endif + + /* Note: all event types must explicitly fill in their lengths here. */ + post_header_len[START_EVENT_V3-1]= START_V3_HEADER_LEN; + post_header_len[QUERY_EVENT-1]= QUERY_HEADER_LEN; + post_header_len[STOP_EVENT-1]= STOP_HEADER_LEN; + post_header_len[ROTATE_EVENT-1]= ROTATE_HEADER_LEN; + post_header_len[INTVAR_EVENT-1]= INTVAR_HEADER_LEN; + post_header_len[LOAD_EVENT-1]= LOAD_HEADER_LEN; + post_header_len[SLAVE_EVENT-1]= SLAVE_HEADER_LEN; + post_header_len[CREATE_FILE_EVENT-1]= CREATE_FILE_HEADER_LEN; + post_header_len[APPEND_BLOCK_EVENT-1]= APPEND_BLOCK_HEADER_LEN; + post_header_len[EXEC_LOAD_EVENT-1]= EXEC_LOAD_HEADER_LEN; + post_header_len[DELETE_FILE_EVENT-1]= DELETE_FILE_HEADER_LEN; + post_header_len[NEW_LOAD_EVENT-1]= NEW_LOAD_HEADER_LEN; + post_header_len[RAND_EVENT-1]= RAND_HEADER_LEN; + post_header_len[USER_VAR_EVENT-1]= USER_VAR_HEADER_LEN; + post_header_len[FORMAT_DESCRIPTION_EVENT-1]= FORMAT_DESCRIPTION_HEADER_LEN; + post_header_len[XID_EVENT-1]= XID_HEADER_LEN; + post_header_len[XA_PREPARE_LOG_EVENT-1]= XA_PREPARE_HEADER_LEN; + post_header_len[BEGIN_LOAD_QUERY_EVENT-1]= BEGIN_LOAD_QUERY_HEADER_LEN; + post_header_len[EXECUTE_LOAD_QUERY_EVENT-1]= EXECUTE_LOAD_QUERY_HEADER_LEN; + /* + The PRE_GA events are never be written to any binlog, but + their lengths are included in Format_description_log_event. + Hence, we need to be assign some value here, to avoid reading + uninitialized memory when the array is written to disk. + */ + post_header_len[PRE_GA_WRITE_ROWS_EVENT-1]= 0; + post_header_len[PRE_GA_UPDATE_ROWS_EVENT-1]= 0; + post_header_len[PRE_GA_DELETE_ROWS_EVENT-1]= 0; + + post_header_len[TABLE_MAP_EVENT-1]= TABLE_MAP_HEADER_LEN; + post_header_len[WRITE_ROWS_EVENT_V1-1]= ROWS_HEADER_LEN_V1; + post_header_len[UPDATE_ROWS_EVENT_V1-1]= ROWS_HEADER_LEN_V1; + post_header_len[DELETE_ROWS_EVENT_V1-1]= ROWS_HEADER_LEN_V1; + /* + We here have the possibility to simulate a master of before we changed + the table map id to be stored in 6 bytes: when it was stored in 4 + bytes (=> post_header_len was 6). This is used to test backward + compatibility. + This code can be removed after a few months (today is Dec 21st 2005), + when we know that the 4-byte masters are not deployed anymore (check + with Tomas Ulin first!), and the accompanying test (rpl_row_4_bytes) + too. + */ + DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master", + post_header_len[TABLE_MAP_EVENT-1]= + post_header_len[WRITE_ROWS_EVENT_V1-1]= + post_header_len[UPDATE_ROWS_EVENT_V1-1]= + post_header_len[DELETE_ROWS_EVENT_V1-1]= 6;); + post_header_len[INCIDENT_EVENT-1]= INCIDENT_HEADER_LEN; + post_header_len[HEARTBEAT_LOG_EVENT-1]= 0; + post_header_len[IGNORABLE_LOG_EVENT-1]= 0; + post_header_len[ROWS_QUERY_LOG_EVENT-1]= 0; + post_header_len[GTID_LOG_EVENT-1]= 0; + post_header_len[ANONYMOUS_GTID_LOG_EVENT-1]= 0; + post_header_len[PREVIOUS_GTIDS_LOG_EVENT-1]= 0; + post_header_len[TRANSACTION_CONTEXT_EVENT-1]= 0; + post_header_len[VIEW_CHANGE_EVENT-1]= 0; + post_header_len[XA_PREPARE_LOG_EVENT-1]= 0; + post_header_len[WRITE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2; + post_header_len[UPDATE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2; + post_header_len[DELETE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2; + + // Set header length of the reserved events to 0 + memset(post_header_len + MYSQL_EVENTS_END - 1, 0, + (MARIA_EVENTS_BEGIN - MYSQL_EVENTS_END)*sizeof(uint8)); + + // Set header lengths of Maria events + post_header_len[ANNOTATE_ROWS_EVENT-1]= ANNOTATE_ROWS_HEADER_LEN; + post_header_len[BINLOG_CHECKPOINT_EVENT-1]= + BINLOG_CHECKPOINT_HEADER_LEN; + post_header_len[GTID_EVENT-1]= GTID_HEADER_LEN; + post_header_len[GTID_LIST_EVENT-1]= GTID_LIST_HEADER_LEN; + post_header_len[START_ENCRYPTION_EVENT-1]= START_ENCRYPTION_HEADER_LEN; + + //compressed event + post_header_len[QUERY_COMPRESSED_EVENT-1]= QUERY_HEADER_LEN; + post_header_len[WRITE_ROWS_COMPRESSED_EVENT-1]= ROWS_HEADER_LEN_V2; + post_header_len[UPDATE_ROWS_COMPRESSED_EVENT-1]= ROWS_HEADER_LEN_V2; + post_header_len[DELETE_ROWS_COMPRESSED_EVENT-1]= ROWS_HEADER_LEN_V2; + post_header_len[WRITE_ROWS_COMPRESSED_EVENT_V1-1]= ROWS_HEADER_LEN_V1; + post_header_len[UPDATE_ROWS_COMPRESSED_EVENT_V1-1]= ROWS_HEADER_LEN_V1; + post_header_len[DELETE_ROWS_COMPRESSED_EVENT_V1-1]= ROWS_HEADER_LEN_V1; + + // Sanity-check that all post header lengths are initialized. + int i; + for (i=0; i<number_of_event_types; i++) + DBUG_ASSERT(post_header_len[i] != 255); + } + break; + + case 1: /* 3.23 */ + case 3: /* 4.0.x x>=2 */ + /* + We build an artificial (i.e. not sent by the master) event, which + describes what those old master versions send. + */ + if (binlog_ver==1) + strmov(server_version, server_ver ? server_ver : "3.23"); + else + strmov(server_version, server_ver ? server_ver : "4.0"); + common_header_len= binlog_ver==1 ? OLD_HEADER_LEN : + LOG_EVENT_MINIMAL_HEADER_LEN; + /* + The first new event in binlog version 4 is Format_desc. So any event type + after that does not exist in older versions. We use the events known by + version 3, even if version 1 had only a subset of them (this is not a + problem: it uses a few bytes for nothing but unifies code; it does not + make the slave detect less corruptions). + */ + number_of_event_types= FORMAT_DESCRIPTION_EVENT - 1; + post_header_len=(uint8*) my_malloc(PSI_INSTRUMENT_ME, + number_of_event_types*sizeof(uint8), MYF(0)); + if (post_header_len) + { + post_header_len[START_EVENT_V3-1]= START_V3_HEADER_LEN; + post_header_len[QUERY_EVENT-1]= QUERY_HEADER_MINIMAL_LEN; + post_header_len[STOP_EVENT-1]= 0; + post_header_len[ROTATE_EVENT-1]= (binlog_ver==1) ? 0 : ROTATE_HEADER_LEN; + post_header_len[INTVAR_EVENT-1]= 0; + post_header_len[LOAD_EVENT-1]= LOAD_HEADER_LEN; + post_header_len[SLAVE_EVENT-1]= 0; + post_header_len[CREATE_FILE_EVENT-1]= CREATE_FILE_HEADER_LEN; + post_header_len[APPEND_BLOCK_EVENT-1]= APPEND_BLOCK_HEADER_LEN; + post_header_len[EXEC_LOAD_EVENT-1]= EXEC_LOAD_HEADER_LEN; + post_header_len[DELETE_FILE_EVENT-1]= DELETE_FILE_HEADER_LEN; + post_header_len[NEW_LOAD_EVENT-1]= post_header_len[LOAD_EVENT-1]; + post_header_len[RAND_EVENT-1]= 0; + post_header_len[USER_VAR_EVENT-1]= 0; + } + break; + default: /* Includes binlog version 2 i.e. 4.0.x x<=1 */ + post_header_len= 0; /* will make is_valid() fail */ + break; + } + calc_server_version_split(); + deduct_options_written_to_bin_log(); + checksum_alg= BINLOG_CHECKSUM_ALG_UNDEF; + reset_crypto(); +} + + +/** + The problem with this constructor is that the fixed header may have a + length different from this version, but we don't know this length as we + have not read the Format_description_log_event which says it, yet. This + length is in the post-header of the event, but we don't know where the + post-header starts. + + So this type of event HAS to: + - either have the header's length at the beginning (in the header, at a + fixed position which will never be changed), not in the post-header. That + would make the header be "shifted" compared to other events. + - or have a header of size LOG_EVENT_MINIMAL_HEADER_LEN (19), in all future + versions, so that we know for sure. + + I (Guilhem) chose the 2nd solution. Rotate has the same constraint (because + it is sent before Format_description_log_event). +*/ + +Format_description_log_event:: +Format_description_log_event(const uchar *buf, uint event_len, + const Format_description_log_event* + description_event) + :Start_log_event_v3(buf, event_len, description_event), + common_header_len(0), post_header_len(NULL), event_type_permutation(0) +{ + DBUG_ENTER("Format_description_log_event::Format_description_log_event(char*,...)"); + if (!Start_log_event_v3::is_valid()) + DBUG_VOID_RETURN; /* sanity check */ + buf+= LOG_EVENT_MINIMAL_HEADER_LEN; + if ((common_header_len=buf[ST_COMMON_HEADER_LEN_OFFSET]) < OLD_HEADER_LEN) + DBUG_VOID_RETURN; /* sanity check */ + number_of_event_types= + event_len - (LOG_EVENT_MINIMAL_HEADER_LEN + ST_COMMON_HEADER_LEN_OFFSET + 1); + DBUG_PRINT("info", ("common_header_len=%d number_of_event_types=%d", + common_header_len, number_of_event_types)); + /* If alloc fails, we'll detect it in is_valid() */ + + post_header_len= (uint8*) my_memdup(PSI_INSTRUMENT_ME, + buf+ST_COMMON_HEADER_LEN_OFFSET+1, + number_of_event_types* + sizeof(*post_header_len), + MYF(0)); + calc_server_version_split(); + if (!is_version_before_checksum(&server_version_split)) + { + /* the last bytes are the checksum alg desc and value (or value's room) */ + number_of_event_types -= BINLOG_CHECKSUM_ALG_DESC_LEN; + checksum_alg= (enum_binlog_checksum_alg)post_header_len[number_of_event_types]; + } + else + { + checksum_alg= BINLOG_CHECKSUM_ALG_UNDEF; + } + deduct_options_written_to_bin_log(); + reset_crypto(); + + DBUG_VOID_RETURN; +} + +bool Format_description_log_event::start_decryption(Start_encryption_log_event* sele) +{ + DBUG_ASSERT(crypto_data.scheme == 0); + + if (!sele->is_valid()) + return 1; + + memcpy(crypto_data.nonce, sele->nonce, BINLOG_NONCE_LENGTH); + return crypto_data.init(sele->crypto_scheme, sele->key_version); +} + + +Version::Version(const char *version, const char **endptr) +{ + 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)) + m_ver[i]= (uchar) number; + else + { + *this= Version(); + break; + } + + p= r; + 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) + kind= KIND_MARIADB; + else + kind= KIND_MYSQL; +} + + +/** + Splits the event's 'server_version' string into three numeric pieces stored + into 'server_version_split': + X.Y.Zabc (X,Y,Z numbers, a not a digit) -> {X,Y,Z} + X.Yabc -> {X,Y,0} + 'server_version_split' is then used for lookups to find if the server which + created this event has some known bug. +*/ +void Format_description_log_event::calc_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[0], + server_version_split[1], server_version_split[2])); +} + + +void Format_description_log_event::deduct_options_written_to_bin_log() +{ + options_written_to_bin_log= OPTION_AUTO_IS_NULL | OPTION_NOT_AUTOCOMMIT | + OPTION_NO_FOREIGN_KEY_CHECKS | OPTION_RELAXED_UNIQUE_CHECKS | + OPTION_INSERT_HISTORY; + if (!server_version_split.version_is_valid() || + server_version_split.kind == master_version_split::KIND_MYSQL || + server_version_split < Version(10,5,2)) + return; + options_written_to_bin_log|= OPTION_IF_EXISTS; + if (server_version_split[0] == 10) + { + const static char v[10]={99,99,99,99,99,17,9,5,4,2}; + if (server_version_split[1] < 10 && + server_version_split[2] < v[server_version_split[1]]) + return; + } + options_written_to_bin_log|= OPTION_EXPLICIT_DEF_TIMESTAMP; + + DBUG_ASSERT(options_written_to_bin_log == OPTIONS_WRITTEN_TO_BIN_LOG); +} + +/** + @return TRUE is the event's version is earlier than one that introduced + the replication event checksum. FALSE otherwise. +*/ +bool +Format_description_log_event::is_version_before_checksum(const master_version_split + *version_split) +{ + return *version_split < + (version_split->kind == master_version_split::KIND_MARIADB ? + checksum_version_split_mariadb : checksum_version_split_mysql); +} + +/** + @param buf buffer holding serialized FD event + @param len netto (possible checksum is stripped off) length of the event buf + + @return the version-safe checksum alg descriptor where zero + designates no checksum, 255 - the orginator is + checksum-unaware (effectively no checksum) and the actuall + [1-254] range alg descriptor. +*/ +enum enum_binlog_checksum_alg get_checksum_alg(const uchar *buf, ulong len) +{ + enum enum_binlog_checksum_alg ret; + char version[ST_SERVER_VER_LEN]; + + DBUG_ENTER("get_checksum_alg"); + DBUG_ASSERT(buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT); + + memcpy(version, + buf + LOG_EVENT_MINIMAL_HEADER_LEN + ST_SERVER_VER_OFFSET, + ST_SERVER_VER_LEN); + version[ST_SERVER_VER_LEN - 1]= 0; + + 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]; + DBUG_ASSERT(ret == BINLOG_CHECKSUM_ALG_OFF || + ret == BINLOG_CHECKSUM_ALG_UNDEF || + ret == BINLOG_CHECKSUM_ALG_CRC32); + DBUG_RETURN(ret); +} + +Start_encryption_log_event:: +Start_encryption_log_event(const uchar *buf, uint event_len, + const Format_description_log_event* description_event) + :Log_event(buf, description_event) +{ + if ((int)event_len == + LOG_EVENT_MINIMAL_HEADER_LEN + Start_encryption_log_event::get_data_size()) + { + buf+= LOG_EVENT_MINIMAL_HEADER_LEN; + crypto_scheme= *buf; + key_version= uint4korr(buf + BINLOG_CRYPTO_SCHEME_LENGTH); + memcpy(nonce, + buf + BINLOG_CRYPTO_SCHEME_LENGTH + BINLOG_KEY_VERSION_LENGTH, + BINLOG_NONCE_LENGTH); + } + else + crypto_scheme= ~0; // invalid +} + + +/************************************************************************** + Load_log_event methods + General note about Load_log_event: the binlogging of LOAD DATA INFILE is + going to be changed in 5.0 (or maybe in 5.1; not decided yet). + However, the 5.0 slave could still have to read such events (from a 4.x + master), convert them (which just means maybe expand the header, when 5.0 + servers have a UID in events) (remember that whatever is after the header + will be like in 4.x, as this event's format is not modified in 5.0 as we + will use new types of events to log the new LOAD DATA INFILE features). + To be able to read/convert, we just need to not assume that the common + header is of length LOG_EVENT_HEADER_LEN (we must use the description + event). + Note that I (Guilhem) manually tested replication of a big LOAD DATA INFILE + between 3.23 and 5.0, and between 4.0 and 5.0, and it works fine (and the + positions displayed in SHOW SLAVE STATUS then are fine too). +**************************************************************************/ + + +/** + @note + The caller must do buf[event_len]= 0 before he starts using the + constructed event. +*/ + +Load_log_event::Load_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + :Log_event(buf, description_event), num_fields(0), fields(0), + field_lens(0),field_block_len(0), + table_name(0), db(0), fname(0), local_fname(FALSE), + /* + Load_log_event which comes from the binary log does not contain + information about the type of insert which was used on the master. + Assume that it was an ordinary, non-concurrent LOAD DATA. + */ + is_concurrent(FALSE) +{ + DBUG_ENTER("Load_log_event"); + /* + I (Guilhem) manually tested replication of LOAD DATA INFILE for 3.23->5.0, + 4.0->5.0 and 5.0->5.0 and it works. + */ + if (event_len) + copy_log_event(buf, event_len, + (((uchar)buf[EVENT_TYPE_OFFSET] == LOAD_EVENT) ? + LOAD_HEADER_LEN + + description_event->common_header_len : + LOAD_HEADER_LEN + LOG_EVENT_HEADER_LEN), + description_event); + /* otherwise it's a derived class, will call copy_log_event() itself */ + DBUG_VOID_RETURN; +} + + +/* + Load_log_event::copy_log_event() +*/ + +int Load_log_event::copy_log_event(const uchar *buf, ulong event_len, + int body_offset, + const Format_description_log_event + *description_event) +{ + DBUG_ENTER("Load_log_event::copy_log_event"); + uint data_len; + if ((int) event_len <= body_offset) + DBUG_RETURN(1); + const uchar *buf_end= buf + event_len; + /* this is the beginning of the post-header */ + const uchar *data_head= buf + description_event->common_header_len; + thread_id= slave_proxy_id= uint4korr(data_head + L_THREAD_ID_OFFSET); + exec_time= uint4korr(data_head + L_EXEC_TIME_OFFSET); + skip_lines= uint4korr(data_head + L_SKIP_LINES_OFFSET); + table_name_len= (uint)data_head[L_TBL_LEN_OFFSET]; + db_len= (uint)data_head[L_DB_LEN_OFFSET]; + num_fields= uint4korr(data_head + L_NUM_FIELDS_OFFSET); + + /* + Sql_ex.init() on success returns the pointer to the first byte after + the sql_ex structure, which is the start of field lengths array. + */ + if (!(field_lens= (uchar*) sql_ex.init(buf + body_offset, buf_end, + buf[EVENT_TYPE_OFFSET] != LOAD_EVENT))) + DBUG_RETURN(1); + + data_len= event_len - body_offset; + if (num_fields > data_len) // simple sanity check against corruption + DBUG_RETURN(1); + for (uint i= 0; i < num_fields; i++) + field_block_len+= (uint)field_lens[i] + 1; + + fields= (char*) field_lens + num_fields; + table_name= fields + field_block_len; + if (strlen(table_name) > NAME_LEN) + goto err; + + db= table_name + table_name_len + 1; + DBUG_EXECUTE_IF("simulate_invalid_address", db_len= data_len;); + fname= db + db_len + 1; + if ((db_len > data_len) || (fname > (char*) buf_end)) + goto err; + fname_len= (uint) strlen(fname); + if ((fname_len > data_len) || (fname + fname_len > (char*) buf_end)) + goto err; + // null termination is accomplished by the caller doing buf[event_len]=0 + + DBUG_RETURN(0); + +err: + // Invalid event. + table_name= 0; + DBUG_RETURN(1); +} + + +/************************************************************************** + Rotate_log_event methods +**************************************************************************/ + +Rotate_log_event::Rotate_log_event(const uchar *buf, uint event_len, + const Format_description_log_event* + description_event) + :Log_event(buf, description_event) ,new_log_ident(0), flags(DUP_NAME) +{ + DBUG_ENTER("Rotate_log_event::Rotate_log_event(char*,...)"); + // The caller will ensure that event_len is what we have at EVENT_LEN_OFFSET + uint8 post_header_len= description_event->post_header_len[ROTATE_EVENT-1]; + uint ident_offset; + if (event_len < (uint)(LOG_EVENT_MINIMAL_HEADER_LEN + post_header_len)) + DBUG_VOID_RETURN; + buf+= LOG_EVENT_MINIMAL_HEADER_LEN; + pos= post_header_len ? uint8korr(buf + R_POS_OFFSET) : 4; + ident_len= (uint)(event_len - (LOG_EVENT_MINIMAL_HEADER_LEN + post_header_len)); + ident_offset= post_header_len; + set_if_smaller(ident_len,FN_REFLEN-1); + new_log_ident= my_strndup(PSI_INSTRUMENT_ME, (char*) buf + ident_offset, + (uint) ident_len, MYF(MY_WME)); + DBUG_PRINT("debug", ("new_log_ident: '%s'", new_log_ident)); + DBUG_VOID_RETURN; +} + + +/************************************************************************** + Binlog_checkpoint_log_event methods +**************************************************************************/ + +Binlog_checkpoint_log_event::Binlog_checkpoint_log_event( + const uchar *buf, uint event_len, + const Format_description_log_event *description_event) + :Log_event(buf, description_event), binlog_file_name(0) +{ + uint8 header_size= description_event->common_header_len; + uint8 post_header_len= + description_event->post_header_len[BINLOG_CHECKPOINT_EVENT-1]; + if (event_len < (uint) header_size + (uint) post_header_len || + post_header_len < BINLOG_CHECKPOINT_HEADER_LEN) + return; + buf+= header_size; + /* See uint4korr and int4store below */ + compile_time_assert(BINLOG_CHECKPOINT_HEADER_LEN == 4); + binlog_file_len= uint4korr(buf); + if (event_len - (header_size + post_header_len) < binlog_file_len) + return; + binlog_file_name= my_strndup(PSI_INSTRUMENT_ME, (char*) buf + post_header_len, + binlog_file_len, MYF(MY_WME)); + return; +} + + +/************************************************************************** + Global transaction ID stuff +**************************************************************************/ + +Gtid_log_event::Gtid_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + : Log_event(buf, description_event), seq_no(0), commit_id(0), + flags_extra(0), extra_engines(0) +{ + uint8 header_size= description_event->common_header_len; + uint8 post_header_len= description_event->post_header_len[GTID_EVENT-1]; + const uchar *buf_0= buf; + if (event_len < (uint) header_size + (uint) post_header_len || + post_header_len < GTID_HEADER_LEN) + return; + + buf+= header_size; + seq_no= uint8korr(buf); + buf+= 8; + domain_id= uint4korr(buf); + buf+= 4; + flags2= *(buf++); + if (flags2 & FL_GROUP_COMMIT_ID) + { + if (event_len < (uint)header_size + GTID_HEADER_LEN + 2) + { + seq_no= 0; // So is_valid() returns false + return; + } + commit_id= uint8korr(buf); + buf+= 8; + } + if (flags2 & (FL_PREPARED_XA | FL_COMPLETED_XA)) + { + xid.formatID= uint4korr(buf); + buf+= 4; + + xid.gtrid_length= (long) buf[0]; + xid.bqual_length= (long) buf[1]; + buf+= 2; + + long data_length= xid.bqual_length + xid.gtrid_length; + memcpy(xid.data, buf, data_length); + buf+= data_length; + } + + /* the extra flags check and actions */ + if (static_cast<uint>(buf - buf_0) < event_len) + { + flags_extra= *buf++; + /* + extra engines flags presence is identifed by non-zero byte value + at this point + */ + if (flags_extra & FL_EXTRA_MULTI_ENGINE_E1) + { + DBUG_ASSERT(static_cast<uint>(buf - buf_0) < event_len); + + extra_engines= *buf++; + + DBUG_ASSERT(extra_engines > 0); + } + if (flags_extra & (FL_COMMIT_ALTER_E1 | FL_ROLLBACK_ALTER_E1)) + { + sa_seq_no= uint8korr(buf); + buf+= 8; + } + } + /* + the strict '<' part of the assert corresponds to extra zero-padded + trailing bytes, + */ + DBUG_ASSERT(static_cast<uint>(buf - buf_0) <= event_len); + /* and the last of them is tested. */ +#ifdef MYSQL_SERVER +#ifdef WITH_WSREP + if (!WSREP_ON) +#endif +#endif + DBUG_ASSERT(static_cast<uint>(buf - buf_0) == event_len || + buf_0[event_len - 1] == 0); +} + +int compare_glle_gtids(const void * _gtid1, const void *_gtid2) +{ + rpl_gtid *gtid1= (rpl_gtid *) _gtid1; + rpl_gtid *gtid2= (rpl_gtid *) _gtid2; + + int ret; + if (*gtid1 < *gtid2) + ret= -1; + else if (*gtid1 > *gtid2) + ret= 1; + else + ret= 0; + return ret; +} + +/* GTID list. */ + +Gtid_list_log_event::Gtid_list_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + : Log_event(buf, description_event), count(0), list(0), sub_id_list(0) +{ + uint32 i; + uint32 val; + uint8 header_size= description_event->common_header_len; + uint8 post_header_len= description_event->post_header_len[GTID_LIST_EVENT-1]; + if (event_len < (uint) header_size + (uint) post_header_len || + post_header_len < GTID_LIST_HEADER_LEN) + return; + + buf+= header_size; + val= uint4korr(buf); + count= val & ((1<<28)-1); + gl_flags= val & ((uint32)0xf << 28); + buf+= 4; + if (event_len - (header_size + post_header_len) < count*element_size || + (!(list= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME, + count*sizeof(*list) + (count == 0), MYF(MY_WME))))) + return; + + for (i= 0; i < count; ++i) + { + list[i].domain_id= uint4korr(buf); + buf+= 4; + list[i].server_id= uint4korr(buf); + buf+= 4; + list[i].seq_no= uint8korr(buf); + buf+= 8; + } + +#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT) + if ((gl_flags & FLAG_IGN_GTIDS)) + { + uint32 i; + if (!(sub_id_list= (uint64 *)my_malloc(PSI_INSTRUMENT_ME, + count*sizeof(uint64), MYF(MY_WME)))) + { + my_free(list); + list= NULL; + return; + } + for (i= 0; i < count; ++i) + { + if (!(sub_id_list[i]= + rpl_global_gtid_slave_state->next_sub_id(list[i].domain_id))) + { + my_free(list); + my_free(sub_id_list); + list= NULL; + sub_id_list= NULL; + return; + } + } + } +#endif +} + + +/* + Used to record gtid_list event while sending binlog to slave, without having to + fully contruct the event object. +*/ +bool +Gtid_list_log_event::peek(const char *event_start, size_t event_len, + enum enum_binlog_checksum_alg checksum_alg, + rpl_gtid **out_gtid_list, uint32 *out_list_len, + const Format_description_log_event *fdev) +{ + const char *p; + uint32 count_field, count; + rpl_gtid *gtid_list; + + if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32) + { + if (event_len > BINLOG_CHECKSUM_LEN) + event_len-= BINLOG_CHECKSUM_LEN; + else + event_len= 0; + } + else + DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF || + checksum_alg == BINLOG_CHECKSUM_ALG_OFF); + + if (event_len < (uint32)fdev->common_header_len + GTID_LIST_HEADER_LEN) + return true; + p= event_start + fdev->common_header_len; + count_field= uint4korr(p); + p+= 4; + count= count_field & ((1<<28)-1); + if (event_len < (uint32)fdev->common_header_len + GTID_LIST_HEADER_LEN + + element_size * count) + return true; + if (!(gtid_list= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME, + sizeof(rpl_gtid)*count + (count == 0), MYF(MY_WME)))) + return true; + *out_gtid_list= gtid_list; + *out_list_len= count; + while (count--) + { + gtid_list->domain_id= uint4korr(p); + p+= 4; + gtid_list->server_id= uint4korr(p); + p+= 4; + gtid_list->seq_no= uint8korr(p); + p+= 8; + ++gtid_list; + } + + return false; +} + + +/************************************************************************** + Intvar_log_event methods +**************************************************************************/ + +/* + Intvar_log_event::Intvar_log_event() +*/ + +Intvar_log_event::Intvar_log_event(const uchar *buf, + const Format_description_log_event* description_event) + :Log_event(buf, description_event) +{ + /* The Post-Header is empty. The Variable Data part begins immediately. */ + buf+= description_event->common_header_len + + description_event->post_header_len[INTVAR_EVENT-1]; + type= buf[I_TYPE_OFFSET]; + val= uint8korr(buf+I_VAL_OFFSET); +} + + +/* + Intvar_log_event::get_var_type_name() +*/ + +const char* Intvar_log_event::get_var_type_name() +{ + switch(type) { + case LAST_INSERT_ID_EVENT: return "LAST_INSERT_ID"; + case INSERT_ID_EVENT: return "INSERT_ID"; + default: /* impossible */ return "UNKNOWN"; + } +} + + +/************************************************************************** + Rand_log_event methods +**************************************************************************/ + +Rand_log_event::Rand_log_event(const uchar *buf, + const Format_description_log_event* description_event) + :Log_event(buf, description_event) +{ + /* The Post-Header is empty. The Variable Data part begins immediately. */ + buf+= description_event->common_header_len + + description_event->post_header_len[RAND_EVENT-1]; + seed1= uint8korr(buf+RAND_SEED1_OFFSET); + seed2= uint8korr(buf+RAND_SEED2_OFFSET); +} + + +/************************************************************************** + Xid_log_event methods +**************************************************************************/ + +/** + @note + It's ok not to use int8store here, + as long as xid_t::set(ulonglong) and + xid_t::get_my_xid doesn't do it either. + We don't care about actual values of xids as long as + identical numbers compare identically +*/ + +Xid_log_event:: +Xid_log_event(const uchar *buf, + const Format_description_log_event *description_event) + :Xid_apply_log_event(buf, description_event) +{ + /* The Post-Header is empty. The Variable Data part begins immediately. */ + buf+= description_event->common_header_len + + description_event->post_header_len[XID_EVENT-1]; + memcpy((char*) &xid, buf, sizeof(xid)); +} + +/************************************************************************** + XA_prepare_log_event methods +**************************************************************************/ +XA_prepare_log_event:: +XA_prepare_log_event(const uchar *buf, + const Format_description_log_event *description_event) + :Xid_apply_log_event(buf, description_event) +{ + buf+= description_event->common_header_len + + description_event->post_header_len[XA_PREPARE_LOG_EVENT-1]; + one_phase= * (bool *) buf; + buf+= 1; + + m_xid.formatID= uint4korr(buf); + buf+= 4; + m_xid.gtrid_length= uint4korr(buf); + buf+= 4; + // Todo: validity here and elsewhere checks to be replaced by MDEV-21839 fixes + if (m_xid.gtrid_length <= 0 || m_xid.gtrid_length > MAXGTRIDSIZE) + { + m_xid.formatID= -1; + return; + } + m_xid.bqual_length= uint4korr(buf); + buf+= 4; + if (m_xid.bqual_length < 0 || m_xid.bqual_length > MAXBQUALSIZE) + { + m_xid.formatID= -1; + return; + } + DBUG_ASSERT(m_xid.gtrid_length + m_xid.bqual_length <= XIDDATASIZE); + + memcpy(m_xid.data, buf, m_xid.gtrid_length + m_xid.bqual_length); + + xid= NULL; +} + + +/************************************************************************** + User_var_log_event methods +**************************************************************************/ + +User_var_log_event:: +User_var_log_event(const uchar *buf, uint event_len, + const Format_description_log_event* description_event) + :Log_event(buf, description_event) +#ifndef MYSQL_CLIENT + , deferred(false), query_id(0) +#endif +{ + bool error= false; + const uchar *buf_start= buf, *buf_end= buf + event_len; + + /* The Post-Header is empty. The Variable Data part begins immediately. */ + buf+= description_event->common_header_len + + description_event->post_header_len[USER_VAR_EVENT-1]; + name_len= uint4korr(buf); + /* Avoid reading out of buffer */ + if ((buf - buf_start) + UV_NAME_LEN_SIZE + name_len > event_len) + { + error= true; + goto err; + } + + name= (char *) buf + UV_NAME_LEN_SIZE; + + /* + We don't know yet is_null value, so we must assume that name_len + may have the bigger value possible, is_null= True and there is no + payload for val, or even that name_len is 0. + */ + if (name + name_len + UV_VAL_IS_NULL > (char*) buf_end) + { + error= true; + goto err; + } + + buf+= UV_NAME_LEN_SIZE + name_len; + is_null= (bool) *buf; + flags= User_var_log_event::UNDEF_F; // defaults to UNDEF_F + if (is_null) + { + type= STRING_RESULT; + charset_number= my_charset_bin.number; + val_len= 0; + val= 0; + } + else + { + val= (char *) (buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE + + UV_CHARSET_NUMBER_SIZE + UV_VAL_LEN_SIZE); + + if (val > (char*) buf_end) + { + error= true; + goto err; + } + + type= (Item_result) buf[UV_VAL_IS_NULL]; + charset_number= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE); + val_len= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE + + UV_CHARSET_NUMBER_SIZE); + + /** + We need to check if this is from an old server + that did not pack information for flags. + We do this by checking if there are extra bytes + after the packed value. If there are we take the + extra byte and it's value is assumed to contain + the flags value. + + Old events will not have this extra byte, thence, + we keep the flags set to UNDEF_F. + */ + size_t bytes_read= (val + val_len) - (char*) buf_start; + if (bytes_read > event_len) + { + error= true; + goto err; + } + if ((data_written - bytes_read) > 0) + { + flags= (uint) *(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE + + UV_CHARSET_NUMBER_SIZE + UV_VAL_LEN_SIZE + + val_len); + } + } + +err: + if (unlikely(error)) + name= 0; +} + + +/************************************************************************** + Create_file_log_event methods +**************************************************************************/ + +/* + Create_file_log_event ctor +*/ + +Create_file_log_event:: +Create_file_log_event(const uchar *buf, uint len, + const Format_description_log_event* description_event) + :Load_log_event(buf,0,description_event),fake_base(0),block(0), + inited_from_old(0) +{ + DBUG_ENTER("Create_file_log_event::Create_file_log_event(char*,...)"); + uint block_offset; + uint header_len= description_event->common_header_len; + uint8 load_header_len= description_event->post_header_len[LOAD_EVENT-1]; + uint8 create_file_header_len= description_event->post_header_len[CREATE_FILE_EVENT-1]; + if (!(event_buf= (uchar*) my_memdup(PSI_INSTRUMENT_ME, buf, len, + MYF(MY_WME))) || + copy_log_event(event_buf,len, + (((uchar)buf[EVENT_TYPE_OFFSET] == LOAD_EVENT) ? + load_header_len + header_len : + (fake_base ? (header_len+load_header_len) : + (header_len+load_header_len) + + create_file_header_len)), + description_event)) + DBUG_VOID_RETURN; + if (description_event->binlog_version!=1) + { + file_id= uint4korr(buf + + header_len + + load_header_len + CF_FILE_ID_OFFSET); + /* + Note that it's ok to use get_data_size() below, because it is computed + with values we have already read from this event (because we called + copy_log_event()); we are not using slave's format info to decode + master's format, we are really using master's format info. + Anyway, both formats should be identical (except the common_header_len) + as these Load events are not changed between 4.0 and 5.0 (as logging of + LOAD DATA INFILE does not use Load_log_event in 5.0). + + The + 1 is for \0 terminating fname + */ + block_offset= (description_event->common_header_len + + Load_log_event::get_data_size() + + create_file_header_len + 1); + if (len < block_offset) + DBUG_VOID_RETURN; + block= const_cast<uchar*>(buf) + block_offset; + block_len= len - block_offset; + } + else + { + sql_ex.force_new_format(); + inited_from_old= 1; + } + DBUG_VOID_RETURN; +} + + +/************************************************************************** + Append_block_log_event methods +**************************************************************************/ + +/* + Append_block_log_event ctor +*/ + +Append_block_log_event:: +Append_block_log_event(const uchar *buf, uint len, + const Format_description_log_event* description_event) + :Log_event(buf, description_event),block(0) +{ + DBUG_ENTER("Append_block_log_event::Append_block_log_event(char*,...)"); + uint8 common_header_len= description_event->common_header_len; + uint8 append_block_header_len= + description_event->post_header_len[APPEND_BLOCK_EVENT-1]; + uint total_header_len= common_header_len+append_block_header_len; + if (len < total_header_len) + DBUG_VOID_RETURN; + file_id= uint4korr(buf + common_header_len + AB_FILE_ID_OFFSET); + block= const_cast<uchar*>(buf) + total_header_len; + block_len= len - total_header_len; + DBUG_VOID_RETURN; +} + + +/************************************************************************** + Delete_file_log_event methods +**************************************************************************/ + +/* + Delete_file_log_event ctor +*/ + +Delete_file_log_event:: +Delete_file_log_event(const uchar *buf, uint len, + const Format_description_log_event* description_event) + :Log_event(buf, description_event),file_id(0) +{ + uint8 common_header_len= description_event->common_header_len; + uint8 delete_file_header_len= description_event->post_header_len[DELETE_FILE_EVENT-1]; + if (len < (uint)(common_header_len + delete_file_header_len)) + return; + file_id= uint4korr(buf + common_header_len + DF_FILE_ID_OFFSET); +} + + +/************************************************************************** + Execute_load_log_event methods +**************************************************************************/ + +/* + Execute_load_log_event ctor +*/ + +Execute_load_log_event:: +Execute_load_log_event(const uchar *buf, uint len, + const Format_description_log_event* description_event) + :Log_event(buf, description_event), file_id(0) +{ + uint8 common_header_len= description_event->common_header_len; + uint8 exec_load_header_len= description_event->post_header_len[EXEC_LOAD_EVENT-1]; + if (len < (uint)(common_header_len+exec_load_header_len)) + return; + file_id= uint4korr(buf + common_header_len + EL_FILE_ID_OFFSET); +} + + +/************************************************************************** + Begin_load_query_log_event methods +**************************************************************************/ + +Begin_load_query_log_event:: +Begin_load_query_log_event(const uchar *buf, uint len, + const Format_description_log_event* desc_event) + :Append_block_log_event(buf, len, desc_event) +{ +} + + +/************************************************************************** + Execute_load_query_log_event methods +**************************************************************************/ + + +Execute_load_query_log_event:: +Execute_load_query_log_event(const uchar *buf, uint event_len, + const Format_description_log_event* desc_event): + Query_log_event(buf, event_len, desc_event, EXECUTE_LOAD_QUERY_EVENT), + file_id(0), fn_pos_start(0), fn_pos_end(0) +{ + if (!Query_log_event::is_valid()) + return; + + buf+= desc_event->common_header_len; + + fn_pos_start= uint4korr(buf + ELQ_FN_POS_START_OFFSET); + fn_pos_end= uint4korr(buf + ELQ_FN_POS_END_OFFSET); + dup_handling= (enum_load_dup_handling)(*(buf + ELQ_DUP_HANDLING_OFFSET)); + + if (fn_pos_start > q_len || fn_pos_end > q_len || + dup_handling > LOAD_DUP_REPLACE) + return; + + file_id= uint4korr(buf + ELQ_FILE_ID_OFFSET); +} + + +ulong Execute_load_query_log_event::get_post_header_size_for_derived() +{ + return EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN; +} + + +/************************************************************************** + sql_ex_info methods +**************************************************************************/ + +/* + sql_ex_info::init() +*/ + +const uchar *sql_ex_info::init(const uchar *buf, const uchar *buf_end, + bool use_new_format) +{ + cached_new_format= use_new_format; + if (use_new_format) + { + empty_flags=0; + /* + The code below assumes that buf will not disappear from + under our feet during the lifetime of the event. This assumption + holds true in the slave thread if the log is in new format, but is not + the case when we have old format because we will be reusing net buffer + to read the actual file before we write out the Create_file event. + */ + if (read_str(&buf, buf_end, &field_term, &field_term_len) || + read_str(&buf, buf_end, &enclosed, &enclosed_len) || + read_str(&buf, buf_end, &line_term, &line_term_len) || + read_str(&buf, buf_end, &line_start, &line_start_len) || + read_str(&buf, buf_end, &escaped, &escaped_len)) + return 0; + opt_flags= *buf++; + } + else + { + if (buf_end - buf < 7) + return 0; // Wrong data + field_term_len= enclosed_len= line_term_len= line_start_len= escaped_len=1; + field_term= (char*) buf++; // Use first byte in string + enclosed= (char*) buf++; + line_term= (char*) buf++; + line_start= (char*) buf++; + escaped= (char*) buf++; + opt_flags= *buf++; + empty_flags= *buf++; + if (empty_flags & FIELD_TERM_EMPTY) + field_term_len=0; + if (empty_flags & ENCLOSED_EMPTY) + enclosed_len=0; + if (empty_flags & LINE_TERM_EMPTY) + line_term_len=0; + if (empty_flags & LINE_START_EMPTY) + line_start_len=0; + if (empty_flags & ESCAPED_EMPTY) + escaped_len=0; + } + return buf; +} + + + +/************************************************************************** + Rows_log_event member functions +**************************************************************************/ + + +Rows_log_event::Rows_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + : Log_event(buf, description_event), + m_row_count(0), +#ifndef MYSQL_CLIENT + m_table(NULL), +#endif + m_table_id(0), m_rows_buf(0), m_rows_cur(0), m_rows_end(0), + m_extra_row_data(0) +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + , m_curr_row(NULL), m_curr_row_end(NULL), + m_key(NULL), m_key_info(NULL), m_key_nr(0), + master_had_triggers(0) +#endif +{ + DBUG_ENTER("Rows_log_event::Rows_log_event(const char*,...)"); + uint8 const common_header_len= description_event->common_header_len; + Log_event_type event_type= (Log_event_type)(uchar)buf[EVENT_TYPE_OFFSET]; + m_type= event_type; + m_cols_ai.bitmap= 0; + + uint8 const post_header_len= description_event->post_header_len[event_type-1]; + + if (event_len < (uint)(common_header_len + post_header_len)) + { + m_cols.bitmap= 0; + DBUG_VOID_RETURN; + } + + DBUG_PRINT("enter",("event_len: %u common_header_len: %d " + "post_header_len: %d", + event_len, common_header_len, + post_header_len)); + + const uchar *post_start= buf + common_header_len; + post_start+= RW_MAPID_OFFSET; + if (post_header_len == 6) + { + /* Master is of an intermediate source tree before 5.1.4. Id is 4 bytes */ + m_table_id= uint4korr(post_start); + post_start+= 4; + } + else + { + m_table_id= (ulong) uint6korr(post_start); + post_start+= RW_FLAGS_OFFSET; + } + + m_flags_pos= post_start - buf; + m_flags= uint2korr(post_start); + post_start+= 2; + + uint16 var_header_len= 0; + if (post_header_len == ROWS_HEADER_LEN_V2) + { + /* + Have variable length header, check length, + which includes length bytes + */ + var_header_len= uint2korr(post_start); + /* Check length and also avoid out of buffer read */ + if (var_header_len < 2 || + event_len < static_cast<unsigned int>(var_header_len + + (post_start - buf))) + { + m_cols.bitmap= 0; + DBUG_VOID_RETURN; + } + var_header_len-= 2; + + /* Iterate over var-len header, extracting 'chunks' */ + const uchar *start= post_start + 2; + const uchar *end= start + var_header_len; + for (const uchar* pos= start; pos < end;) + { + switch(*pos++) + { + case RW_V_EXTRAINFO_TAG: + { + /* Have an 'extra info' section, read it in */ + assert((end - pos) >= EXTRA_ROW_INFO_HDR_BYTES); + uint8 infoLen= pos[EXTRA_ROW_INFO_LEN_OFFSET]; + assert((end - pos) >= infoLen); + /* Just store/use the first tag of this type, skip others */ + if (likely(!m_extra_row_data)) + { + m_extra_row_data= (uchar*) my_malloc(PSI_INSTRUMENT_ME, infoLen, + MYF(MY_WME)); + if (likely(m_extra_row_data != NULL)) + { + memcpy(m_extra_row_data, pos, infoLen); + } + } + pos+= infoLen; + break; + } + default: + /* Unknown code, we will not understand anything further here */ + pos= end; /* Break loop */ + } + } + } + + uchar const *const var_start= + (const uchar *)buf + common_header_len + post_header_len + var_header_len; + uchar const *const ptr_width= var_start; + uchar *ptr_after_width= (uchar*) ptr_width; + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + m_width= net_field_length(&ptr_after_width); + DBUG_PRINT("debug", ("m_width=%lu", m_width)); + + /* Avoid reading out of buffer */ + if (ptr_after_width + (m_width + 7) / 8 > (uchar*)buf + event_len) + { + m_cols.bitmap= NULL; + DBUG_VOID_RETURN; + } + + /* if my_bitmap_init fails, caught in is_valid() */ + if (likely(!my_bitmap_init(&m_cols, + m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL, + m_width))) + { + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + memcpy(m_cols.bitmap, ptr_after_width, (m_width + 7) / 8); + create_last_word_mask(&m_cols); + ptr_after_width+= (m_width + 7) / 8; + DBUG_DUMP("m_cols", (uchar*) m_cols.bitmap, no_bytes_in_map(&m_cols)); + } + else + { + // Needed because my_bitmap_init() does not set it to null on failure + m_cols.bitmap= NULL; + DBUG_VOID_RETURN; + } + + m_cols_ai.bitmap= m_cols.bitmap; /* See explanation in is_valid() */ + + if (LOG_EVENT_IS_UPDATE_ROW(event_type)) + { + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + + /* if my_bitmap_init fails, caught in is_valid() */ + if (likely(!my_bitmap_init(&m_cols_ai, + m_width <= sizeof(m_bitbuf_ai)*8 ? m_bitbuf_ai : NULL, + m_width))) + { + DBUG_PRINT("debug", ("Reading from %p", ptr_after_width)); + memcpy(m_cols_ai.bitmap, ptr_after_width, (m_width + 7) / 8); + create_last_word_mask(&m_cols_ai); + ptr_after_width+= (m_width + 7) / 8; + DBUG_DUMP("m_cols_ai", (uchar*) m_cols_ai.bitmap, + no_bytes_in_map(&m_cols_ai)); + } + else + { + // Needed because my_bitmap_init() does not set it to null on failure + m_cols_ai.bitmap= 0; + DBUG_VOID_RETURN; + } + } + + const uchar* const ptr_rows_data= (const uchar*) ptr_after_width; + + size_t const read_size= ptr_rows_data - (const unsigned char *) buf; + if (read_size > event_len) + { + DBUG_VOID_RETURN; + } + size_t const data_size= event_len - read_size; + DBUG_PRINT("info",("m_table_id: %llu m_flags: %d m_width: %lu data_size: %lu", + m_table_id, m_flags, m_width, (ulong) data_size)); + + m_rows_buf= (uchar*) my_malloc(PSI_INSTRUMENT_ME, data_size, MYF(MY_WME)); + if (likely((bool)m_rows_buf)) + { +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + m_curr_row= m_rows_buf; +#endif + m_rows_end= m_rows_buf + data_size; + m_rows_cur= m_rows_end; + memcpy(m_rows_buf, ptr_rows_data, data_size); + m_rows_before_size= ptr_rows_data - (const uchar *) buf; // Get the size that before SET part + } + else + m_cols.bitmap= 0; // to not free it + + DBUG_VOID_RETURN; +} + +void Rows_log_event::uncompress_buf() +{ + uint32 un_len= binlog_get_uncompress_len(m_rows_buf); + if (!un_len) + return; + + uchar *new_buf= (uchar*) my_malloc(PSI_INSTRUMENT_ME, ALIGN_SIZE(un_len), + MYF(MY_WME)); + if (new_buf) + { + if (!binlog_buf_uncompress(m_rows_buf, new_buf, + (uint32)(m_rows_cur - m_rows_buf), &un_len)) + { + my_free(m_rows_buf); + m_rows_buf= new_buf; +#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION) + m_curr_row= m_rows_buf; +#endif + m_rows_end= m_rows_buf + un_len; + m_rows_cur= m_rows_end; + return; + } + else + { + my_free(new_buf); + } + } + m_cols.bitmap= 0; // catch it in is_valid +} + +Rows_log_event::~Rows_log_event() +{ + if (m_cols.bitmap == m_bitbuf) // no my_malloc happened + m_cols.bitmap= 0; // so no my_free in my_bitmap_free + my_bitmap_free(&m_cols); // To pair with my_bitmap_init(). + my_free(m_rows_buf); + my_free(m_extra_row_data); +} + +int Rows_log_event::get_data_size() +{ + int const general_type_code= get_general_type_code(); + + uchar buf[MAX_INT_WIDTH]; + uchar *end= net_store_length(buf, m_width); + + DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master", + return (int)(6 + no_bytes_in_map(&m_cols) + (end - buf) + + (general_type_code == UPDATE_ROWS_EVENT ? no_bytes_in_map(&m_cols_ai) : 0) + + m_rows_cur - m_rows_buf);); + int data_size= 0; + Log_event_type type= get_type_code(); + bool is_v2_event= LOG_EVENT_IS_ROW_V2(type); + if (is_v2_event) + { + data_size= ROWS_HEADER_LEN_V2 + + (m_extra_row_data ? + RW_V_TAG_LEN + m_extra_row_data[EXTRA_ROW_INFO_LEN_OFFSET]: + 0); + } + else + { + data_size= ROWS_HEADER_LEN_V1; + } + data_size+= no_bytes_in_map(&m_cols); + data_size+= (uint) (end - buf); + + if (general_type_code == UPDATE_ROWS_EVENT) + data_size+= no_bytes_in_map(&m_cols_ai); + + data_size+= (uint) (m_rows_cur - m_rows_buf); + return data_size; +} + + +/************************************************************************** + Annotate_rows_log_event member functions +**************************************************************************/ + +Annotate_rows_log_event:: +Annotate_rows_log_event(const uchar *buf, + uint event_len, + const Format_description_log_event *desc) + : Log_event(buf, desc), + m_save_thd_query_txt(0), + m_save_thd_query_len(0), + m_saved_thd_query(false), + m_used_query_txt(0) +{ + m_query_len= event_len - desc->common_header_len; + m_query_txt= (char*) buf + desc->common_header_len; +} + +Annotate_rows_log_event::~Annotate_rows_log_event() +{ + DBUG_ENTER("Annotate_rows_log_event::~Annotate_rows_log_event"); +#ifndef MYSQL_CLIENT + if (m_saved_thd_query) + thd->set_query(m_save_thd_query_txt, m_save_thd_query_len); + else if (m_used_query_txt) + thd->reset_query(); +#endif + DBUG_VOID_RETURN; +} + +int Annotate_rows_log_event::get_data_size() +{ + return m_query_len; +} + +Log_event_type Annotate_rows_log_event::get_type_code() +{ + return ANNOTATE_ROWS_EVENT; +} + +bool Annotate_rows_log_event::is_valid() const +{ + return (m_query_txt != NULL && m_query_len != 0); +} + + +/************************************************************************** + Table_map_log_event member functions and support functions +**************************************************************************/ + +/** + @page How replication of field metadata works. + + When a table map is created, the master first calls + Table_map_log_event::save_field_metadata() which calculates how many + values will be in the field metadata. Only those fields that require the + extra data are added. The method also loops through all of the fields in + the table calling the method Field::save_field_metadata() which returns the + values for the field that will be saved in the metadata and replicated to + the slave. Once all fields have been processed, the table map is written to + the binlog adding the size of the field metadata and the field metadata to + the end of the body of the table map. + + When a table map is read on the slave, the field metadata is read from the + table map and passed to the table_def class constructor which saves the + field metadata from the table map into an array based on the type of the + field. Field metadata values not present (those fields that do not use extra + data) in the table map are initialized as zero (0). The array size is the + same as the columns for the table on the slave. + + Additionally, values saved for field metadata on the master are saved as a + string of bytes (uchar) in the binlog. A field may require 1 or more bytes + to store the information. In cases where values require multiple bytes + (e.g. values > 255), the endian-safe methods are used to properly encode + the values on the master and decode them on the slave. When the field + metadata values are captured on the slave, they are stored in an array of + type uint16. This allows the least number of casts to prevent casting bugs + when the field metadata is used in comparisons of field attributes. When + the field metadata is used for calculating addresses in pointer math, the + type used is uint32. +*/ + +/* + Constructor used by slave to read the event from the binary log. + */ +#if defined(HAVE_REPLICATION) +Table_map_log_event::Table_map_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + + : Log_event(buf, description_event), +#ifndef MYSQL_CLIENT + m_table(NULL), +#endif + m_dbnam(NULL), m_dblen(0), m_tblnam(NULL), m_tbllen(0), + m_colcnt(0), m_coltype(0), + m_memory(NULL), m_table_id(ULONGLONG_MAX), m_flags(0), + m_data_size(0), m_field_metadata(0), m_field_metadata_size(0), + m_null_bits(0), m_meta_memory(NULL), + m_optional_metadata_len(0), m_optional_metadata(NULL) +{ + unsigned int bytes_read= 0; + DBUG_ENTER("Table_map_log_event::Table_map_log_event(const char*,uint,...)"); + + uint8 common_header_len= description_event->common_header_len; + uint8 post_header_len= description_event->post_header_len[TABLE_MAP_EVENT-1]; + DBUG_PRINT("info",("event_len: %u common_header_len: %d post_header_len: %d", + event_len, common_header_len, post_header_len)); + + /* + Don't print debug messages when running valgrind since they can + trigger false warnings. + */ +#ifndef HAVE_valgrind + DBUG_DUMP("event buffer", (uchar*) buf, event_len); +#endif + + if (event_len < (uint)(common_header_len + post_header_len)) + DBUG_VOID_RETURN; + + /* Read the post-header */ + const uchar *post_start= buf + common_header_len; + + post_start+= TM_MAPID_OFFSET; + VALIDATE_BYTES_READ(post_start, buf, event_len); + if (post_header_len == 6) + { + /* Master is of an intermediate source tree before 5.1.4. Id is 4 bytes */ + m_table_id= uint4korr(post_start); + post_start+= 4; + } + else + { + DBUG_ASSERT(post_header_len == TABLE_MAP_HEADER_LEN); + m_table_id= (ulong) uint6korr(post_start); + post_start+= TM_FLAGS_OFFSET; + } + + DBUG_ASSERT(m_table_id != ~0ULL); + + m_flags= uint2korr(post_start); + + /* Read the variable part of the event */ + const uchar *const vpart= buf + common_header_len + post_header_len; + + /* Extract the length of the various parts from the buffer */ + uchar const *const ptr_dblen= (uchar const*)vpart + 0; + VALIDATE_BYTES_READ(ptr_dblen, buf, event_len); + m_dblen= *(uchar*) ptr_dblen; + + /* Length of database name + counter + terminating null */ + uchar const *const ptr_tbllen= ptr_dblen + m_dblen + 2; + VALIDATE_BYTES_READ(ptr_tbllen, buf, event_len); + m_tbllen= *(uchar*) ptr_tbllen; + + /* Length of table name + counter + terminating null */ + uchar const *const ptr_colcnt= ptr_tbllen + m_tbllen + 2; + uchar *ptr_after_colcnt= (uchar*) ptr_colcnt; + VALIDATE_BYTES_READ(ptr_after_colcnt, buf, event_len); + m_colcnt= net_field_length(&ptr_after_colcnt); + + DBUG_PRINT("info",("m_dblen: %lu off: %ld m_tbllen: %lu off: %ld m_colcnt: %lu off: %ld", + (ulong) m_dblen, (long) (ptr_dblen - vpart), + (ulong) m_tbllen, (long) (ptr_tbllen - vpart), + m_colcnt, (long) (ptr_colcnt - vpart))); + + /* Allocate mem for all fields in one go. If fails, caught in is_valid() */ + m_memory= (uchar*) my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME), + &m_dbnam, (uint) m_dblen + 1, + &m_tblnam, (uint) m_tbllen + 1, + &m_coltype, (uint) m_colcnt, + NullS); + + if (m_memory) + { + /* Copy the different parts into their memory */ + strncpy(const_cast<char*>(m_dbnam), (const char*)ptr_dblen + 1, m_dblen + 1); + strncpy(const_cast<char*>(m_tblnam), (const char*)ptr_tbllen + 1, m_tbllen + 1); + memcpy(m_coltype, ptr_after_colcnt, m_colcnt); + + ptr_after_colcnt= ptr_after_colcnt + m_colcnt; + VALIDATE_BYTES_READ(ptr_after_colcnt, buf, event_len); + m_field_metadata_size= net_field_length(&ptr_after_colcnt); + if (m_field_metadata_size <= (m_colcnt * 2)) + { + uint num_null_bytes= (m_colcnt + 7) / 8; + m_meta_memory= (uchar *)my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME), + &m_null_bits, num_null_bytes, + &m_field_metadata, m_field_metadata_size, + NULL); + memcpy(m_field_metadata, ptr_after_colcnt, m_field_metadata_size); + ptr_after_colcnt= (uchar*)ptr_after_colcnt + m_field_metadata_size; + memcpy(m_null_bits, ptr_after_colcnt, num_null_bytes); + ptr_after_colcnt= (unsigned char*)ptr_after_colcnt + num_null_bytes; + } + else + { + m_coltype= NULL; + my_free(m_memory); + m_memory= NULL; + DBUG_VOID_RETURN; + } + + bytes_read= (uint) (ptr_after_colcnt - (uchar *)buf); + + /* After null_bits field, there are some new fields for extra metadata. */ + if (bytes_read < event_len) + { + m_optional_metadata_len= event_len - bytes_read; + m_optional_metadata= + static_cast<unsigned char*>(my_malloc(PSI_INSTRUMENT_ME, m_optional_metadata_len, MYF(MY_WME))); + memcpy(m_optional_metadata, ptr_after_colcnt, m_optional_metadata_len); + } + } +#ifdef MYSQL_SERVER + if (!m_table) + DBUG_VOID_RETURN; + binlog_type_info_array= (Binlog_type_info *)thd->alloc(m_table->s->fields * + sizeof(Binlog_type_info)); + for (uint i= 0; i < m_table->s->fields; i++) + binlog_type_info_array[i]= m_table->field[i]->binlog_type_info(); +#endif + + DBUG_VOID_RETURN; +} +#endif + +Table_map_log_event::~Table_map_log_event() +{ + my_free(m_meta_memory); + my_free(m_memory); + my_free(m_optional_metadata); + m_optional_metadata= NULL; +} + +/** + Parses SIGNEDNESS field. + + @param[out] vec stores the signedness flags extracted from field. + @param[in] field SIGNEDNESS field in table_map_event. + @param[in] length length of the field + */ +static void parse_signedness(std::vector<bool> &vec, + unsigned char *field, unsigned int length) +{ + for (unsigned int i= 0; i < length; i++) + { + for (unsigned char c= 0x80; c != 0; c>>= 1) + vec.push_back(field[i] & c); + } +} + +/** + Parses DEFAULT_CHARSET field. + + @param[out] default_charset stores collation numbers extracted from field. + @param[in] field DEFAULT_CHARSET field in table_map_event. + @param[in] length length of the field + */ +static void parse_default_charset(Table_map_log_event::Optional_metadata_fields:: + Default_charset &default_charset, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + default_charset.default_charset= net_field_length(&p); + while (p < field + length) + { + unsigned int col_index= net_field_length(&p); + unsigned int col_charset= net_field_length(&p); + + default_charset.charset_pairs.push_back(std::make_pair(col_index, + col_charset)); + } +} + +/** + Parses COLUMN_CHARSET field. + + @param[out] vec stores collation numbers extracted from field. + @param[in] field COLUMN_CHARSET field in table_map_event. + @param[in] length length of the field + */ +static void parse_column_charset(std::vector<unsigned int> &vec, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + while (p < field + length) + vec.push_back(net_field_length(&p)); +} + +/** + Parses COLUMN_NAME field. + + @param[out] vec stores column names extracted from field. + @param[in] field COLUMN_NAME field in table_map_event. + @param[in] length length of the field + */ +static void parse_column_name(std::vector<std::string> &vec, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + while (p < field + length) + { + unsigned len= net_field_length(&p); + vec.push_back(std::string(reinterpret_cast<char *>(p), len)); + p+= len; + } +} + +/** + Parses SET_STR_VALUE/ENUM_STR_VALUE field. + + @param[out] vec stores SET/ENUM column's string values extracted from + field. Each SET/ENUM column's string values are stored + into a string separate vector. All of them are stored + in 'vec'. + @param[in] field COLUMN_NAME field in table_map_event. + @param[in] length length of the field + */ +static void parse_set_str_value(std::vector<Table_map_log_event:: + Optional_metadata_fields::str_vector> &vec, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + while (p < field + length) + { + unsigned int count= net_field_length(&p); + + vec.push_back(std::vector<std::string>()); + for (unsigned int i= 0; i < count; i++) + { + unsigned len1= net_field_length(&p); + vec.back().push_back(std::string(reinterpret_cast<char *>(p), len1)); + p+= len1; + } + } +} + +/** + Parses GEOMETRY_TYPE field. + + @param[out] vec stores geometry column's types extracted from field. + @param[in] field GEOMETRY_TYPE field in table_map_event. + @param[in] length length of the field + */ +static void parse_geometry_type(std::vector<unsigned int> &vec, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + while (p < field + length) + vec.push_back(net_field_length(&p)); +} + +/** + Parses SIMPLE_PRIMARY_KEY field. + + @param[out] vec stores primary key's column information extracted from + field. Each column has an index and a prefix which are + stored as a unit_pair. prefix is always 0 for + SIMPLE_PRIMARY_KEY field. + @param[in] field SIMPLE_PRIMARY_KEY field in table_map_event. + @param[in] length length of the field + */ +static void parse_simple_pk(std::vector<Table_map_log_event:: + Optional_metadata_fields::uint_pair> &vec, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + while (p < field + length) + vec.push_back(std::make_pair(net_field_length(&p), 0)); +} + +/** + Parses PRIMARY_KEY_WITH_PREFIX field. + + @param[out] vec stores primary key's column information extracted from + field. Each column has an index and a prefix which are + stored as a unit_pair. + @param[in] field PRIMARY_KEY_WITH_PREFIX field in table_map_event. + @param[in] length length of the field + */ + +static void parse_pk_with_prefix(std::vector<Table_map_log_event:: + Optional_metadata_fields::uint_pair> &vec, + unsigned char *field, unsigned int length) +{ + unsigned char* p= field; + + while (p < field + length) + { + unsigned int col_index= net_field_length(&p); + unsigned int col_prefix= net_field_length(&p); + vec.push_back(std::make_pair(col_index, col_prefix)); + } +} + +Table_map_log_event::Optional_metadata_fields:: +Optional_metadata_fields(unsigned char* optional_metadata, + unsigned int optional_metadata_len) +{ + unsigned char* field= optional_metadata; + + if (optional_metadata == NULL) + return; + + while (field < optional_metadata + optional_metadata_len) + { + unsigned int len; + Optional_metadata_field_type type= + static_cast<Optional_metadata_field_type>(field[0]); + + // Get length and move field to the value. + field++; + len= net_field_length(&field); + + switch(type) + { + case SIGNEDNESS: + parse_signedness(m_signedness, field, len); + break; + case DEFAULT_CHARSET: + parse_default_charset(m_default_charset, field, len); + break; + case COLUMN_CHARSET: + parse_column_charset(m_column_charset, field, len); + break; + case COLUMN_NAME: + parse_column_name(m_column_name, field, len); + break; + case SET_STR_VALUE: + parse_set_str_value(m_set_str_value, field, len); + break; + case ENUM_STR_VALUE: + parse_set_str_value(m_enum_str_value, field, len); + break; + case GEOMETRY_TYPE: + parse_geometry_type(m_geometry_type, field, len); + break; + case SIMPLE_PRIMARY_KEY: + parse_simple_pk(m_primary_key, field, len); + break; + case PRIMARY_KEY_WITH_PREFIX: + parse_pk_with_prefix(m_primary_key, field, len); + break; + case ENUM_AND_SET_DEFAULT_CHARSET: + parse_default_charset(m_enum_and_set_default_charset, field, len); + break; + case ENUM_AND_SET_COLUMN_CHARSET: + parse_column_charset(m_enum_and_set_column_charset, field, len); + break; + default: + DBUG_ASSERT(0); + } + // next field + field+= len; + } +} + + +/************************************************************************** + Write_rows_log_event member functions +**************************************************************************/ + + +/* + Constructor used by slave to read the event from the binary log. + */ +#ifdef HAVE_REPLICATION +Write_rows_log_event::Write_rows_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) +: Rows_log_event(buf, event_len, description_event) +{ +} + +Write_rows_compressed_log_event::Write_rows_compressed_log_event( + const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) +: Write_rows_log_event(buf, event_len, description_event) +{ + uncompress_buf(); +} +#endif + + +/************************************************************************** + Delete_rows_log_event member functions +**************************************************************************/ + +/* + Constructor used by slave to read the event from the binary log. + */ +#ifdef HAVE_REPLICATION +Delete_rows_log_event::Delete_rows_log_event(const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + : Rows_log_event(buf, event_len, description_event) +{ +} + +Delete_rows_compressed_log_event::Delete_rows_compressed_log_event( + const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + : Delete_rows_log_event(buf, event_len, description_event) +{ + uncompress_buf(); +} +#endif + +/************************************************************************** + Update_rows_log_event member functions +**************************************************************************/ + +Update_rows_log_event::~Update_rows_log_event() +{ + if (m_cols_ai.bitmap) + { + if (m_cols_ai.bitmap == m_bitbuf_ai) // no my_malloc happened + m_cols_ai.bitmap= 0; // so no my_free in my_bitmap_free + my_bitmap_free(&m_cols_ai); // To pair with my_bitmap_init(). + } +} + + +/* + Constructor used by slave to read the event from the binary log. + */ +#ifdef HAVE_REPLICATION +Update_rows_log_event::Update_rows_log_event(const uchar *buf, uint event_len, + const + Format_description_log_event + *description_event) + : Rows_log_event(buf, event_len, description_event) +{ +} + +Update_rows_compressed_log_event::Update_rows_compressed_log_event( + const uchar *buf, uint event_len, + const Format_description_log_event + *description_event) + : Update_rows_log_event(buf, event_len, description_event) +{ + uncompress_buf(); +} +#endif + +Incident_log_event::Incident_log_event(const uchar *buf, uint event_len, + const Format_description_log_event *descr_event) + : Log_event(buf, descr_event) +{ + DBUG_ENTER("Incident_log_event::Incident_log_event"); + uint8 const common_header_len= + descr_event->common_header_len; + uint8 const post_header_len= + descr_event->post_header_len[INCIDENT_EVENT-1]; + + DBUG_PRINT("info",("event_len: %u; common_header_len: %d; post_header_len: %d", + event_len, common_header_len, post_header_len)); + + m_message.str= NULL; + m_message.length= 0; + int incident_number= uint2korr(buf + common_header_len); + if (incident_number >= INCIDENT_COUNT || + incident_number <= INCIDENT_NONE) + { + // If the incident is not recognized, this binlog event is + // invalid. If we set incident_number to INCIDENT_NONE, the + // invalidity will be detected by is_valid(). + m_incident= INCIDENT_NONE; + DBUG_VOID_RETURN; + } + m_incident= static_cast<Incident>(incident_number); + uchar const *ptr= buf + common_header_len + post_header_len; + uchar const *const str_end= buf + event_len; + uint8 len= 0; // Assignment to keep compiler happy + const char *str= NULL; // Assignment to keep compiler happy + if (read_str(&ptr, str_end, &str, &len)) + { + /* Mark this event invalid */ + m_incident= INCIDENT_NONE; + DBUG_VOID_RETURN; + } + if (!(m_message.str= (char*) my_malloc(key_memory_log_event, len+1, MYF(MY_WME)))) + { + /* Mark this event invalid */ + m_incident= INCIDENT_NONE; + DBUG_VOID_RETURN; + } + strmake(m_message.str, str, len); + m_message.length= len; + DBUG_PRINT("info", ("m_incident: %d", m_incident)); + DBUG_VOID_RETURN; +} + + +Incident_log_event::~Incident_log_event() +{ + if (m_message.str) + my_free(m_message.str); +} + + +const char * +Incident_log_event::description() const +{ + static const char *const description[]= { + "NOTHING", // Not used + "LOST_EVENTS" + }; + + DBUG_PRINT("info", ("m_incident: %d", m_incident)); + return description[m_incident]; +} + + +Ignorable_log_event::Ignorable_log_event(const uchar *buf, + const Format_description_log_event + *descr_event, + const char *event_name) + :Log_event(buf, descr_event), number((int) (uchar) buf[EVENT_TYPE_OFFSET]), + description(event_name) +{ + DBUG_ENTER("Ignorable_log_event::Ignorable_log_event"); + DBUG_VOID_RETURN; +} + +Ignorable_log_event::~Ignorable_log_event() = default; + +bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, FILE *file) +{ + return (my_b_copy_all_to_file(cache, file) || + reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE)); +} + +#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) +int Log_event::apply_event(rpl_group_info* rgi) +{ + int res; + THD_STAGE_INFO(thd, stage_apply_event); + rgi->current_event= this; + res= do_apply_event(rgi); + rgi->current_event= NULL; + THD_STAGE_INFO(thd, stage_after_apply_event); + return res; +} +#endif |