| Quellcodebibliothek Statistik Leitseite products/Sources/formale Sprachen/C/MariaDB/sql/ (MariaDB Server Version 8.1-8.4©) Datei vom 1.9.2026 mit Größe 146 kB |
|
Impressum log_event.cc Sprache: C |
|||||||||||||||
|
/* 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; /** 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= CREATE_TYPELIB_FOR(binlog_checksum_type_names); #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= NUL : 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 }; #if !defined DBUG_OFF && defined DBUG_TRACE #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: 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 uchar *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 + 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; } /* Write the length of an event no replica can uncompress, so that a replica's handling of one can be tested. Reaching a value this large takes four length bytes, and the encoding above spends four only on content of 16MB or more, so the injection sets the encoding as well as the value. The compressed content still goes where those four bytes place it, leaving the event well formed apart from the length it claims. */ DBUG_EXECUTE_IF("binlog_compress_corrupt_len", { dst[1]= 0xFF; dst[2]= 0xFF; dst[3]= 0xFF; dst[4]= 0xFC; lenlen= 4; }); 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 returned as false, then 'dst' reuses the passed-in 'buf'. return zero if successful, otherwise the error code the caller reports. */ 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 (unlikely(src_len < len)) return ER_BINLOG_UNCOMPRESS_ERROR; 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 (unlikely(end <= tmp)) return ER_BINLOG_UNCOMPRESS_ERROR; 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 (unlikely(end <= tmp)) return ER_BINLOG_UNCOMPRESS_ERROR; int32 comp_len= (int32)(len - (tmp - src) - (contain_checksum ? BINLOG_CHECKSUM_LEN : 0)); uint32 un_len= binlog_get_uncompress_len(tmp); // bad event if (unlikely(comp_len < 0 || un_len == 0)) return ER_BINLOG_UNCOMPRESS_ERROR; if (unlikely(un_len > MAX_MAX_ALLOWED_PACKET)) return ER_TOO_BIG_FOR_UNCOMPRESS; *newlen= (ulong)(tmp - src) + un_len; if (contain_checksum) *newlen+= BINLOG_CHECKSUM_LEN; size_t alloc_size= 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 ER_BINLOG_UNCOMPRESS_ERROR; *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 ER_BINLOG_UNCOMPRESS_ERROR; } 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 (unlikely(src_len < len)) return ER_BINLOG_UNCOMPRESS_ERROR; 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 (unlikely(end - tmp <= 2)) return ER_BINLOG_UNCOMPRESS_ERROR; 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 (unlikely(end <= tmp)) return ER_BINLOG_UNCOMPRESS_ERROR; 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 (unlikely(end <= tmp)) return ER_BINLOG_UNCOMPRESS_ERROR; uint32 un_len= binlog_get_uncompress_len(tmp); if (unlikely(un_len == 0)) return ER_BINLOG_UNCOMPRESS_ERROR; if (unlikely(un_len > MAX_MAX_ALLOWED_PACKET)) return ER_TOO_BIG_FOR_UNCOMPRESS; int32 comp_len= (int32)(len - (tmp - src) - (contain_checksum ? BINLOG_CHECKSUM_LEN : 0)); if (unlikely(comp_len <= 0)) return ER_BINLOG_UNCOMPRESS_ERROR; *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 ER_BINLOG_UNCOMPRESS_ERROR; *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 ER_BINLOG_UNCOMPRESS_ERROR; } 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 PARTIAL_UPDATE_ROWS_EVENT: return "MySQL Update_rows_partial"; case TRANSACTION_PAYLOAD_EVENT: return "MySQL Transaction_payload"; case HEARTBEAT_LOG_EVENT_V2: return "MySQL Heartbeat"; 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"; case PARTIAL_ROW_DATA_EVENT: return "Partial_rows"; 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) #ifndef MYSQL_CLIENT , slave_exec_mode(SLAVE_EXEC_MODE_STRICT) #endif { #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); log_pos= uint4korr(buf + LOG_POS_OFFSET); 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_binlog_checksum_alg checksum_alg_arg, size_t max_allowed_packet) { ulong data_len; char buf[LOG_EVENT_MINIMAL_HEADER_LEN]; uchar ev_offset= packet->length(); 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= (uint) sz - ev_offset - 4; 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" 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, int *out_error, const Format_description_log_event *fdle, my_bool crc_check, my_bool print_errors, size_t max_allowed_packet) { 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; *out_error= 0; switch (read_log_event(file, &event, fdle, BINLOG_CHECKSUM_ALG_OFF, max_allowed_packet)) { 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; } /* print_errors is false to prevent redundant error messages cluttering up the log, as it will be printed below (if _our_ print_errors is true) */ if ((res= read_log_event((uchar*) event.ptr(), event.length(), &error, fdle, crc_check, false))) 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 /* The SQL slave thread will check *out_error 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. */ *out_error= 1; /* Clear any error that might have been set in the IO_CACHE from a read error, while we are still holding the relay log mutex (if reading from the hot log). Otherwise the error might interfere unpredictably with write operations to the same IO_CACHE in the IO thread. */ file->error= 0; #ifndef MYSQL_CLIENT if (!print_errors) DBUG_RETURN(res); #endif if (event.length() >= LOG_EVENT_MINIMAL_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); } DBUG_RETURN(res); } /** Binlog format tolerance is in (buf, event_len, fdle) constructors. */ Log_event* Log_event::read_log_event(const uchar *buf, size_t event_len, const char **error, const Format_description_log_event *fdle, my_bool crc_check, my_bool print_errors) { Log_event* ev= nullptr; 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); *error= 0; /* 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]; switch (event_type) { case FORMAT_DESCRIPTION_EVENT: // If event is FD the descriptor is in it. if (unlikely(get_checksum_alg(buf, event_len, &alg))) { #ifdef MYSQL_CLIENT if (force_opt) { event_len-= BINLOG_CHECKSUM_LEN; ev= new Unknown_log_event(buf, fdle); goto exit; } #endif *error= "Found invalid event in binary log"; DBUG_RETURN(nullptr); } break; case START_EVENT_V3: // all following START events in the current file are without checksum (const_cast< Format_description_log_event *>(fdle))->used_checksum_alg= BINLOG_CHECKSUM_ALG_OFF; // fall-through default: /* 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. 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= fdle->used_checksum_alg; // Emulate the corruption during reading an event DBUG_EXECUTE_IF("corrupt_read_log_event_char", { 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 DBUG_SET("-d,corrupt_read_log_event_char"); } ); } if (crc_check && event_checksum_test(const_cast<uchar*>(buf), event_len, alg)) { #ifdef MYSQL_CLIENT if (force_opt) { event_len-= BINLOG_CHECKSUM_LEN; ev= new Unknown_log_event(buf, fdle); goto exit; } *error= "Event crc check failed! Most likely there is event corruption."; #else *error= ER_THD_OR_DEFAULT(current_thd, ER_BINLOG_READ_EVENT_CHECKSUM_FAILURE); if (print_errors) sql_print_error("%s", *error); #endif DBUG_RETURN(NULL); } 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; ev= Log_event::read_log_event_no_checksum(buf, event_len, error, fdle); } #ifdef MYSQL_CLIENT exit: if (ev) { ev->read_checksum_alg= alg; if (alg != BINLOG_CHECKSUM_ALG_OFF && alg != BINLOG_CHECKSUM_ALG_UNDEF) ev->read_checksum_value= uint4korr(buf + (event_len)); } #endif DBUG_RETURN(ev); } Log_event *Log_event::read_log_event_no_checksum( const uchar *buf, size_t event_len, const char **error, const Format_description_log_event *fdle) { Log_event* ev; DBUG_ENTER("Log_event::read_log_event_no_checksum(char*,...)"); DBUG_ASSERT(fdle != 0); DBUG_PRINT("info", ("binlog_version: %d", fdle->binlog_version)); DBUG_DUMP_EVENT_BUF(buf, event_len); *error= 0; /* 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]; /* 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; } /* 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: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Query_log_event(buf, static_cast<uint>(event_len), fdle, QUERY_EVENT); break; case QUERY_COMPRESSED_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Query_compressed_log_event(buf, static_cast<uint>(event_len), fdle, QUERY_COMPRESSED_EVENT); break; case ROTATE_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Rotate_log_event(buf, static_cast<uint>(event_len), fdle); break; case BINLOG_CHECKPOINT_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Binlog_checkpoint_log_event(buf, static_cast<uint>(event_len), fdle); break; case GTID_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Gtid_log_event(buf, static_cast<uint>(event_len), fdle); break; case GTID_LIST_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Gtid_list_log_event(buf, static_cast<uint>(event_len), fdle); break; case APPEND_BLOCK_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Append_block_log_event(buf, static_cast<uint>(event_len), fdle); break; case DELETE_FILE_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Delete_file_log_event(buf, static_cast<uint>(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: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new User_var_log_event(buf, static_cast<uint>(event_len), fdle); break; case FORMAT_DESCRIPTION_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Format_description_log_event(buf, static_cast<uint>(event_len), fdle); break; #if defined(HAVE_REPLICATION) 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 HEARTBEAT_LOG_EVENT_V2: // MySQL 8.0 case VIEW_CHANGE_EVENT: ev= new Ignorable_log_event(buf, fdle, get_type_str((Log_event_type) event_type)); break; case TABLE_MAP_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Table_map_log_event(buf, static_cast<uint>(event_len), fdle); break; #endif case PARTIAL_ROW_DATA_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Partial_rows_log_event(buf, static_cast<uint>(event_len), fdle); break; case BEGIN_LOAD_QUERY_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Begin_load_query_log_event(buf, static_cast<uint>(event_len), fdle); break; case EXECUTE_LOAD_QUERY_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Execute_load_query_log_event(buf, static_cast<uint>(event_len), fdle); break; case INCIDENT_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Incident_log_event(buf, static_cast<uint>(event_len), fdle); break; case ANNOTATE_ROWS_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Annotate_rows_log_event(buf, static_cast<uint>(event_len), fdle); break; case START_ENCRYPTION_EVENT: DBUG_ASSERT(event_len <= UINT32_MAX); ev= new Start_encryption_log_event(buf, static_cast<uint>(event_len), fdle); break; case TRANSACTION_PAYLOAD_EVENT: // MySQL 8.0 *error= "Found incompatible MySQL 8.0 TRANSACTION_PAYLOAD_EVENT event. " "You can avoid this event by specifying " "'binlog_transaction_compression=0' in the MySQL server"; ev= NULL; break; case PARTIAL_UPDATE_ROWS_EVENT: // MySQL 8.0 *error= "Found incompatible MySQL 8.0 PARTIAL_UPDATE_ROWS_EVENT event. " "You can avoid this event by specifying " "'binlog-row-value-options=\"\"' in the MySQL server"; ev= NULL; break; case PRE_GA_WRITE_ROWS_EVENT: case PRE_GA_UPDATE_ROWS_EVENT: case PRE_GA_DELETE_ROWS_EVENT: case START_EVENT_V3: /* this is sent only by MySQL <=4.x */ case CREATE_FILE_EVENT: case EXEC_LOAD_EVENT: case LOAD_EVENT: case NEW_LOAD_EVENT: default: DBUG_PRINT("error",("Unknown event code: %d", (uchar) buf[EVENT_TYPE_OFFSET])); ev= NULL; break; } exit: DBUG_PRINT("read_event", ("%s(type_code: %u; event_len: %zu)", 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 */ { if (!*error) *error= "Found invalid event in binary log"; DBUG_RETURN(0); } ev= new Unknown_log_event(buf, fdle); #else if (!*error) *error= "Found invalid event in binary log"; DBUG_RETURN(0); #endif } DBUG_RETURN(ev); } int handler_binlog_reader::read_log_event(String *packet, uint32_t ev_offset, size_t max_allowed) { uint32_t sofar= 0; bool header_read= false; uint32_t target_size= EVENT_LEN_OFFSET + 4; int res; if (unlikely(!buf)) return LOG_READ_MEM; /* Loop, first reading the "length" field, and then continuing to read data until a full event has been placed in the packet. */ for (;;) { if (buf_data_remain <= 0) { res= read_binlog_data(buf, BUF_SIZE); if (res <= 0) { res= (res < 0 ? LOG_READ_IO : LOG_READ_EOF); goto err; } buf_data_pos= 0; buf_data_remain= res; } uint32_t amount= std::min(target_size - sofar, buf_data_remain); packet->append((char *)buf + buf_data_pos, amount); buf_data_pos+= amount; buf_data_remain-= amount; sofar+= amount; if (target_size == sofar) { if (header_read) break; else { header_read= true; target_size= uint4korr(&((*packet)[EVENT_LEN_OFFSET + ev_offset])); if (target_size < LOG_EVENT_MINIMAL_HEADER_LEN) { res= LOG_READ_BOGUS; goto err; } else if (target_size > max_allowed) { res= LOG_READ_TOO_LARGE; goto err; } /* Note that here we rely on the fact that all valid events have more data after the length. This way we avoid conditional for the (useless) special case where we don't need to read anything more after having read the first part. */ DBUG_ASSERT(LOG_EVENT_MINIMAL_HEADER_LEN > EVENT_LEN_OFFSET+4); } } } res= 0; /* Success */ err: return res; } /* 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"; case Q_CHARACTER_SET_COLLATIONS: return "Q_CHARACTER_SET_COLLATIONS"; } snprintf(buf, sizeof(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), character_set_collations({0,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), gtid_flags_extra(0), sa_seq_no(0) { ulong data_len; 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); 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)); /* 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_CHARACTER_SET_COLLATIONS: { const uchar *pos0= pos; CHECK_SPACE(pos, end, 1); uint16 count= *pos++; CHECK_SPACE(pos, end, count * 4); pos+= count * 4; character_set_collations= Lex_cstring((const char *) pos0, (const char *) pos); 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: // impossible CHECK_SPACE(pos, end, 4); 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; } case Q_DUMMY: { /* At some point, this query event was translated from a GTID event, with these Q_DUMMY bytes added to pad the end of the header. We can skip the rest of processing these vars. Note this is a separate case from the default to avoid the DBUG_PRINT of an unknown status var. */ pos= (const uchar*) end; 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 | uninitialized 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) 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) size_t db_length= (size_t)db_len; memcpy(start + data_len + 1, &db_length, sizeof(size_t)); #endif DBUG_VOID_RETURN; } /* Get the time when the event had been executed on the master. This works for both query events and load data events. */ #if Q_EXEC_TIME_OFFSET != L_EXEC_TIME_OFFSET #error "Q_EXEC_TIME_OFFSET is not same as L_EXEC_TIME_OFFSET" #endif time_t query_event_get_end_time(const uchar *buf, const Format_description_log_event *description_event) { time_t when; DBUG_ASSERT(LOG_EVENT_IS_QUERY((Log_event_type) buf[EVENT_TYPE_OFFSET]) || LOG_EVENT_IS_LOAD_DATA((Log_event_type) buf[EVENT_TYPE_OFFSET])); when= uint4korr(buf); buf+= description_event->common_header_len; return when + uint4korr(buf + Q_EXEC_TIME_OFFSET); } 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 (unlikely(!un_len || un_len > MAX_MAX_ALLOWED_PACKET)) { 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_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_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; uint dummy_bytes; 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); 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 the allocated GTID event packet header is longer than the size of the standard BEGIN query event's, then we need to fill in everything else with "dummy" values. That is, old slaves won't recognize the meaning for the DUMMY value, and will skip the rest of the status vars section. */ DBUG_ASSERT(data_len >= LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN); if (data_len < LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN) return 1; DBUG_ASSERT(std::numeric_limits<uint16>::max() >= (data_len - (LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN))); dummy_bytes= static_cast<uint>(data_len - (LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN)); int2store(q + Q_STATUS_VARS_LEN_OFFSET, dummy_bytes); bfill(&q[Q_DATA_OFFSET], dummy_bytes, Q_DUMMY); q[Q_DATA_OFFSET + dummy_bytes]= 0; /* Zero terminator for empty db */ q+= Q_DATA_OFFSET + dummy_bytes + 1; 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; } /*************************************************************************** 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, enum_binlog_checksum_alg checksum_alg) :Log_event(), created(0), binlog_version(binlog_ver), dont_set_created(0), event_type_permutation(0), used_checksum_alg(checksum_alg) { switch (binlog_version) { 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[PARTIAL_UPDATE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2; post_header_len[TRANSACTION_PAYLOAD_EVENT-1]= ROWS_HEADER_LEN_V2; post_header_len[HEARTBEAT_LOG_EVENT_V2-1]= ROWS_HEADER_LEN_V2; 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; post_header_len[PARTIAL_ROW_DATA_EVENT-1]= PARTIAL_ROWS_HEADER_LEN; // 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 */ 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(); 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) :Log_event(buf, description_event), binlog_version(BINLOG_VERSION), common_header_len(0), post_header_len(NULL), event_type_permutation(0) { DBUG_ENTER("Format_description_log_event::Format_description_log_event(char*,... used_checksum_alg= BINLOG_CHECKSUM_ALG_UNDEF; if (unlikely( event_len < LOG_EVENT_MINIMAL_HEADER_LEN + ST_POST_HEADER_LEN_OFFSET)) { server_version[0]= 0; DBUG_VOID_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; if (unlikely(server_version[0] == 0)) DBUG_VOID_RETURN; /* sanity check */ if (unlikely( (common_header_len=buf[ST_COMMON_HEADER_LEN_OFFSET]) < LOG_EVENT_MINIMAL_HEADER_LEN DBUG_VOID_RETURN; /* sanity check */ number_of_event_types= event_len - (LOG_EVENT_MINIMAL_HEADER_LEN + ST_POST_HEADER_LEN_OFFSET); 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() */ calc_server_version_split(); buf+= ST_POST_HEADER_LEN_OFFSET; if (!is_version_before_checksum(&server_version_split)) { /* the last bytes are the checksum alg desc */ if (unlikely(number_of_event_types < BINLOG_CHECKSUM_ALG_DESC_LEN)) DBUG_VOID_RETURN; /* sanity check: But there are no last bytes. */ number_of_event_types -= BINLOG_CHECKSUM_ALG_DESC_LEN; used_checksum_alg= (enum_binlog_checksum_alg)buf[number_of_event_types]; } else { used_checksum_alg= BINLOG_CHECKSUM_ALG_OFF; } post_header_len= (uint8*) my_memdup(PSI_INSTRUMENT_ME, buf, number_of_event_types* sizeof(*post_header_len), MYF(0)); deduct_options_written_to_bin_log(); reset_crypto(); DBUG_VOID_RETURN; } bool Format_description_log_event::start_decryption(Start_encryption_log_event* se { 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 | OPTION_NO_CHECK_CONSTRAINT_CHECKS; 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_spl *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 including the 4-byte checksum @param len length of the event buf @param alg output the version-safe checksum alg descriptor where zero designates no checksum, 255 - the orginator is checksum-unaware (effectively no checksum) and the actual [1-254] range alg descriptor. @return whether this is an invalid FD event */ bool get_checksum_alg(const uchar *buf, size_t len, enum_binlog_checksum_alg *alg) { constexpr ptrdiff_t POST_HEADER_LEN_OFFSET= LOG_EVENT_MINIMAL_HEADER_LEN + ST_POST_HEADER_LEN_OFFSET; char version[ST_SERVER_VER_LEN]; DBUG_ENTER("get_checksum_alg"); DBUG_ASSERT(buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT); if (unlikely(len < POST_HEADER_LEN_OFFSET)) DBUG_RETURN(true); 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); if (Format_description_log_event::is_version_before_checksum(&version_split)) *alg= BINLOG_CHECKSUM_ALG_UNDEF; else { /* len >= POST_HEADER_LEN_OFFSET > BINLOG_CHECKSUM_LEN + BINLOG_CHECKSUM_ALG_DESC_LEN */ size_t checksum_alg_offset= len - BINLOG_CHECKSUM_LEN - BINLOG_CHECKSUM_ALG_DESC_LEN; if (unlikely(checksum_alg_offset < POST_HEADER_LEN_OFFSET)) DBUG_RETURN(true); *alg= static_cast<enum_binlog_checksum_alg>(buf[checksum_alg_offset]); } DBUG_ASSERT(*alg == BINLOG_CHECKSUM_ALG_OFF || *alg == BINLOG_CHECKSUM_ALG_UNDEF || *alg == BINLOG_CHECKSUM_ALG_CRC32); DBUG_RETURN(false); } 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 } /************************************************************************** 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), thread_id(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)) { if (event_len < static_cast<uint>(buf - buf_0) + 6) { seq_no= 0; return; } 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; if (event_len < static_cast<uint>(buf - buf_0) + data_length || xid.gtrid_length > MAXGTRIDSIZE || xid.bqual_length > MAXBQUALSIZE) { xid.formatID= -1; seq_no= 0; return; } 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 identified by non-zero byte value at this point */ if (flags_extra & FL_EXTRA_MULTI_ENGINE_E1) { if (event_len < static_cast<uint>(buf - buf_0) + 1) { seq_no= 0; return; } extra_engines= *buf++; DBUG_ASSERT(extra_engines > 0); } if (flags_extra & (FL_COMMIT_ALTER_E1 | FL_ROLLBACK_ALTER_E1)) { if (event_len < static_cast<uint>(buf - buf_0) + 8) { seq_no= 0; return; } sa_seq_no= uint8korr(buf); buf+= 8; } if (flags_extra & FL_EXTRA_THREAD_ID && static_cast<uint>(buf - buf_0) <= event_len + 4) { thread_id= uint4korr(buf); buf+= 4; } } /* 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 construct the event object. */ bool Gtid_list_log_event::peek(const char *event_start, size_t event_len, 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 **************************************************************************/ bool Log_event_data_type::unpack_optional_attributes(const char *pos, const char *end) { for ( ; pos < end; ) { switch (*pos) { case CHUNK_SIGNED: m_is_unsigned= false; pos++; continue; case CHUNK_UNSIGNED: m_is_unsigned= true; pos++; continue; case CHUNK_DATA_TYPE_NAME: { pos++; if (pos >= end) return true; uint length= (uchar) *pos++; if (pos + length > end) return true; m_data_type_name= {pos, length}; pos+= length; continue; } default: break; // Unknown chunk } } return false; } 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 *const buf_start= buf; const char *buf_end= reinterpret_cast<const char*>(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; if (is_null) { 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; } m_type= (Item_result) buf[UV_VAL_IS_NULL]; m_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 m_is_unsigned==false. */ const char *pos= val + val_len; if (pos > buf_end || unpack_optional_attributes(pos, buf_end)) { error= true; goto err; } } err: if (unlikely(error)) name= 0; } /************************************************************************** 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- if (len < (uint)(common_header_len + delete_file_header_len)) return; file_id= uint4korr(buf + common_header_len + DF_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, size_t 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), m_usable_key_parts(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; // Set to invalid, so it can be processed in is_valid(). uint8 const post_header_len= description_event->post_header_len[event_type-1]; if (event_len < (uint)(common_header_len + post_header_len)) DBUG_VOID_RETURN; DBUG_PRINT("enter",("event_len: %zu 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= (ulonglong) 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 */ if (unlikely((end - pos) <= EXTRA_ROW_INFO_LEN_OFFSET)) { m_cols.bitmap= 0; DBUG_VOID_RETURN; } uint8 infoLen= pos[EXTRA_ROW_INFO_LEN_OFFSET]; if (unlikely(infoLen < EXTRA_ROW_INFO_HDR_BYTES || (end-pos) < infoLen)) { m_cols.bitmap= 0; DBUG_VOID_RETURN; } /* 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=%u", 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)); bitmap_import(&m_cols, ptr_after_width); DBUG_DUMP("m_cols", (uchar*) ptr_after_width, no_bytes_in_export_map(&m_cols)); ptr_after_width+= (m_width + 7) / 8; } else DBUG_VOID_RETURN; 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)); bitmap_import(&m_cols_ai, ptr_after_width); DBUG_DUMP("m_cols_ai", ptr_after_width, no_bytes_in_export_map(&m_cols_ai)); ptr_after_width+= (m_width + 7) / 8; } else { DBUG_ASSERT(m_cols_ai.bitmap == NULL); DBUG_VOID_RETURN; } } else { m_cols_ai= m_cols; /* Safety */ #ifdef DBUG_OFF /* m_cols_ai should only be usable for update events. Make sure nobody successfully manipulates it in debug builds. */ m_cols_ai.bitmap= (my_bitmap_map*)1; #endif } 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: %u 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 (unlikely(!un_len || un_len > MAX_MAX_ALLOWED_PACKET)) { /* my_bitmap_free() nulls m_cols.bitmap, which is_valid() rejects. */ my_bitmap_free(&m_cols); 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); } } my_bitmap_free(&m_cols); // catch it in is_valid } Rows_log_event::~Rows_log_event() { 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_export_map(&m_cols) + (end - buf) + (general_type_code == UPDATE_ROWS_EVENT ? no_bytes_in_export_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_export_map(&m_cols); data_size+= (uint) (end - buf); if (general_type_code == UPDATE_ROWS_EVENT) data_size+= no_bytes_in_export_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= (ulonglong) uint6korr(post_start); post_start+= TM_FLAGS_OFFSET; } DBUG_ASSERT((m_table_id & MAX_TABLE_MAP_ID) != UINT32_MAX && (m_table_id & MAX_TABLE_MAP_ID) != 0); 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; /* If either the database or table name is too long, return before allocating memory for the event, so future validity checks on this event (i.e. is_valid()) return false. */ if (unlikely(m_dblen > NAME_LEN || m_tbllen > NAME_LEN)) DBUG_VOID_RETURN; /* 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_EXECUTE_IF("corrupt_table_map_colcnt_read", m_colcnt= (1 << 20);); 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); /* Future uses of the database and table name require a null-terminating string, but the source string from the above copy may not have one, and strncpy will not add one. So always add it to ensure it exists. */ const_cast<char*>(m_dbnam)[m_dblen]= '\0'; const_cast<char*>(m_tblnam)[m_tbllen]= '\0'; if (unlikely(ptr_after_colcnt + m_colcnt > buf + event_len)) { my_free(m_memory); m_memory= NULL; DBUG_VOID_RETURN; } 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); DBUG_EXECUTE_IF("corrupt_table_map_field_metadata_size_read", m_field_metadata_size= (1 << 20);); /* Ensure the field metadata block covers the metadata the declared column types consume. */ size_t metadata_needed= 0; for (ulong i= 0; i < m_colcnt; i++) metadata_needed+= table_def::field_metadata_length(m_coltype[i]); if (unlikely(m_field_metadata_size < metadata_needed)) { m_coltype= NULL; my_free(m_memory); m_memory= NULL; DBUG_VOID_RETURN; } 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); if (unlikely(ptr_after_colcnt + m_field_metadata_size > buf + event_len)) { my_free(m_meta_memory); m_meta_memory= NULL; my_free(m_memory); m_memory= NULL; DBUG_VOID_RETURN; } memcpy(m_field_metadata, ptr_after_colcnt, m_field_metadata_size); ptr_after_colcnt= (uchar*)ptr_after_colcnt + m_field_metadata_size; if (unlikely(ptr_after_colcnt + num_null_bytes > buf + event_len)) { my_free(m_meta_memory); m_meta_memory= NULL; my_free(m_memory); m_memory= NULL; DBUG_VOID_RETURN; } 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 memcpy(m_optional_metadata, ptr_after_colcnt, m_optional_metadata_len); } } #ifdef MYSQL_SERVER if (!m_table) DBUG_VOID_RETURN; binlog_type_info_array= thd->alloc<Binlog_type_info>(m_table->s->fields); 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] column_metadata store all column metadata here (including signedness) @param[in] field SIGNEDNESS field in table_map_event. @param[in] length length of the field */ using Optional_column_metadata= Table_map_log_event::Optional_metadata_fields::Col static void parse_signedness( Dynamic_array<Optional_column_metadata>& column_metadata, unsigned char *field, unsigned int length) { unsigned char *p= field; unsigned char *field_end= p + length; unsigned char unsigned_bitfield= 0; unsigned char mask= 0; for (uint col= 0; col < column_metadata.size(); col++) { auto &col_meta= column_metadata.at(col); if (!is_numeric_type(col_meta.column_type)) continue; if (mask == 0) { if (p >= field_end) return; unsigned_bitfield= *p++; mask= 0x80; } col_meta.is_unsigned= unsigned_bitfield & mask; mask >>= 1; } } /** Parses DEFAULT_CHARSET field. @param[out] fields store all optional metadata here (including charsets) @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& fields, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; uint default_cs= net_field_length(&p); if (unlikely(p > end)) return; for (uint i= 0; i < fields.m_column_metadata.size(); i++) if (is_character_type(fields.m_column_metadata.at(i).column_type)) fields.m_column_metadata.at(i).charset= default_cs; while (p < end) { unsigned int col_index= net_field_length(&p); if (unlikely(p > end)) return; uint charset= net_field_length(&p); if (unlikely(p > end)) return; if (col_index < fields.m_column_metadata.size()) fields.m_column_metadata.at(col_index).charset= charset; } } /** Parses ENUM_AND_SET_DEFAULT_CHARSET field. @param[out] fields store all optional metadata here (including charsets) @param[in] field ENUM_AND_SET_DEFAULT_CHARSET field in table_map_event. @param[in] length length of the field */ static void parse_enum_and_set_default_charset( Table_map_log_event::Optional_metadata_fields& fields, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; uint default_cs= net_field_length(&p); if (unlikely(p > end)) return; for (uint i= 0; i < fields.m_column_metadata.size(); i++) if (is_enum_or_set_type(fields.m_column_metadata.at(i).column_type)) fields.m_column_metadata.at(i).enum_and_set_column_charset= default_cs; while (p < end) { unsigned int col_index= net_field_length(&p); if (unlikely(p > end)) return; uint charset= net_field_length(&p); if (unlikely(p > end)) return; if (col_index < fields.m_column_metadata.size()) fields.m_column_metadata.at(col_index).enum_and_set_column_charset= charset; } } /** Parses COLUMN_CHARSET field. @param[out] column_metadata store all column metadata here (including charsets) @param[in] field COLUMN_CHARSET field in table_map_event. @param[in] length length of the field */ static void parse_column_charset( Dynamic_array<Optional_column_metadata>& column_metadata, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; for (uint col= 0; p < end && col < column_metadata.size(); col++) { if (!is_character_type(column_metadata.at(col).column_type)) continue; column_metadata.at(col).charset= net_field_length(&p); if (unlikely(p > end)) return; } } /** Parses ENUM_AND_SET_COLUMN_CHARSET field. @param[out] column_metadata store all column metadata here (including charsets) @param[in] field ENUM_AND_SET_COLUMN_CHARSET field in table_map_event. @param[in] length length of the field */ static void parse_enum_and_set_column_charset( Dynamic_array<Optional_column_metadata>& column_metadata, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; for (uint col= 0; p < end && col < column_metadata.size(); col++) { if (!is_enum_or_set_type(column_metadata.at(col).column_type)) continue; column_metadata.at(col).enum_and_set_column_charset= net_field_length(&p); if (unlikely(p > end)) return; } } /** Parses COLUMN_NAME field. @param[in] root Allocate memory here @param[out] column_metadata store all column metadata here (including column names) @param[in] field COLUMN_NAME field in table_map_event. @param[in] length length of the field */ static bool parse_column_name(MEM_ROOT *root, Dynamic_array<Optional_column_metadata>& column_metadata, unsigned char *field, unsigned int length) { unsigned int col = 0; for (uchar *end= field+length; field < end && col < column_metadata.size(); col++) { LEX_CSTRING& name = column_metadata.at(col).column_name; uint name_length= net_field_length(&field); if (unlikely(field + name_length > end)) return 1; if (!(name.str= strmake_root(root, (char*) field, name_length))) return 1; name.length= name_length; field+= name_length; } return 0; } /** Parses SET_STR_VALUE/ENUM_STR_VALUE field. @param[out] column_metadata store all column metadata here (including set/enum strings) @param[in] field SET_STR_VALUE/ENUM_STR_VALUE field in table_map_event. @param[in] length length of the field @param[in] column_type type of the given column (i.e. MYSQL_TYPE_XYZ) @param[in] var_to_set pointer-to-member selecting which str_vector to fill @retval false ok @retval true out of memory while reserving the str_vector */ static bool parse_set_str_value( Dynamic_array<Optional_column_metadata> &column_metadata, unsigned char *field, unsigned int length, uchar column_type, Table_map_log_event::Optional_metadata_fields::str_vector Optional_column_metadata::*var_to_set) { unsigned char* p= field; unsigned char* end= field + length; for (uint col= 0; p < end && col < column_metadata.size(); col++) { if (column_metadata.at(col).column_type != column_type) continue; unsigned int count= net_field_length(&p); if (unlikely(p > end)) return 0; auto* column_strings= &(column_metadata.at(col).*var_to_set); if (column_strings->reserve(count)) return 1; for (unsigned int i= 0; i < count; i++) { unsigned len1= net_field_length(&p); if (unlikely(p + len1 > end)) return 0; column_strings->append(LEX_CSTRING{reinterpret_cast<char *>(p), len1}); p+= len1; } } return 0; } /** Parses GEOMETRY_TYPE field. @param[out] column_metadata store all column metadata here (including geometry types) @param[in] field GEOMETRY_TYPE field in table_map_event. @param[in] length length of the field */ static void parse_geometry_type( Dynamic_array<Optional_column_metadata> &column_metadata, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; for (unsigned int col = 0; p < end && col < column_metadata.size(); col++) { auto& col_metadata = column_metadata.at(col); if (col_metadata.column_type == MYSQL_TYPE_GEOMETRY) { col_metadata.geometry_type = net_field_length(&p); if (unlikely(p > end)) return; } } } /** Parses SIMPLE_PRIMARY_KEY field. @param[out] column_metadata store all column metadata here (including primary key information). The prefix is always 0 for SIMPLE_PRIMARY_KEY. @param[in] field SIMPLE_PRIMARY_KEY field in table_map_event. @param[in] length length of the field */ static void parse_simple_pk( Dynamic_array<Optional_column_metadata> &column_metadata, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; while (p < end) { unsigned int col_index= net_field_length(&p); if (unlikely(p > end)) return; if (col_index < column_metadata.size()) column_metadata.at(col_index).primary_key = 0; } } /** Parses PRIMARY_KEY_WITH_PREFIX field. @param[out] column_metadata store all column metadata here (including primary key lengths) @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( Dynamic_array<Optional_column_metadata> &column_metadata, unsigned char *field, unsigned int length) { unsigned char* p= field; unsigned char* end= field + length; while (p < end) { unsigned int col_index= net_field_length(&p); if (unlikely(p > end)) return; unsigned int col_prefix= net_field_length(&p); if (unlikely(p > end)) return; if (col_index < column_metadata.size()) column_metadata.at(col_index).primary_key = col_prefix; } } Table_map_log_event::Optional_metadata_fields:: Optional_metadata_fields(MEM_ROOT *root, uint master_columns, const uchar* column_types, const uchar* field_metadata, uchar* optional_metadata, size_t optional_metadata_len, bool only_column_names) : m_column_metadata(PSI_INSTRUMENT_MEM, master_columns) { unsigned int len; uchar *metadata_end; allocation_error= 0; // initializing the elements of m_column_metadata and their types for (uint i= 0; i < master_columns; i++) { Column_metadata col{}; col.column_type = column_types[i]; if (field_metadata) { if (col.column_type == MYSQL_TYPE_STRING && (*field_metadata == MYSQL_TYPE_ENUM || *field_metadata == MYSQL_TYPE_SET)) col.column_type= *field_metadata; advance_field_metadata_ptr(col.column_type, &field_metadata); } if (m_column_metadata.append(col)) goto error; } if (optional_metadata == NULL) return; metadata_end= optional_metadata + optional_metadata_len; for (uchar *field= optional_metadata ; field < metadata_end ; field+= 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); if (only_column_names && type != COLUMN_NAME) continue; switch(type) { case SIGNEDNESS: parse_signedness(m_column_metadata, field, len); break; case DEFAULT_CHARSET: parse_default_charset(*this, field, len); break; case COLUMN_CHARSET: parse_column_charset(m_column_metadata, field, len); break; case COLUMN_NAME: if (parse_column_name(root, m_column_metadata, field, len)) goto error; break; case SET_STR_VALUE: if (parse_set_str_value(m_column_metadata, field, len, MYSQL_TYPE_SET, &Optional_column_metadata::set_str_values)) goto error; break; case ENUM_STR_VALUE: if (parse_set_str_value(m_column_metadata, field, len, MYSQL_TYPE_ENUM, &Optional_column_metadata::enum_str_values)) goto error; break; case GEOMETRY_TYPE: parse_geometry_type(m_column_metadata, field, len); break; case SIMPLE_PRIMARY_KEY: parse_simple_pk(m_column_metadata, field, len); break; case PRIMARY_KEY_WITH_PREFIX: parse_pk_with_prefix(m_column_metadata, field, len); break; case ENUM_AND_SET_DEFAULT_CHARSET: parse_enum_and_set_default_charset(*this, field, len); break; case ENUM_AND_SET_COLUMN_CHARSET: parse_enum_and_set_column_charset(m_column_metadata, field, len); break; default: DBUG_ASSERT(0); } } return; error: allocation_error= 1; /* The column names may be partially filled if parsing failed midway. Clear all of them so callers checking the first column's name fall back to positional column mapping instead of using a half-parsed name list. */ for (size_t i= 0; i < m_column_metadata.size(); i++) m_column_metadata.at(i).column_name= LEX_CSTRING{nullptr, 0}; return; } /************************************************************************** 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, size_t 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, size_t 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, size_t 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, size_t 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() { 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, size_t 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, size_t 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; } Partial_rows_log_event::Partial_rows_log_event( const uchar *buf, uint event_len, const Format_description_log_event *description_event) : Log_event(buf, description_event), metadata_written(0), rows_event(NULL) { DBUG_ENTER("Partial_rows_log_event::Partial_rows_log_even(const uchar*,uint,...)" uint8 common_header_len= description_event->common_header_len; uint8 post_header_len= description_event->post_header_len[PARTIAL_ROW_DATA_EVENT-1] DBUG_PRINT("info",("event_len: %u common_header_len: %d post_header_len: %d", event_len, common_header_len, post_header_len)); DBUG_ASSERT(post_header_len == PARTIAL_ROWS_HEADER_LEN); if (event_len < (uint)(common_header_len + post_header_len)) DBUG_VOID_RETURN; start_offset= common_header_len + PARTIAL_ROWS_HEADER_LEN; ev_buffer_base= buf; end_offset= event_len; /* Read the post-header */ const uchar *post_start= buf + common_header_len; VALIDATE_BYTES_READ(post_start, buf, event_len); total_fragments= uint4korr(post_start); post_start+= 4; VALIDATE_BYTES_READ(post_start, buf, event_len); seq_no= uint4korr((post_start)); post_start+= 4; VALIDATE_BYTES_READ(post_start, buf, event_len); DBUG_ASSERT(seq_no <= total_fragments); flags2= *(post_start++); VALIDATE_BYTES_READ(post_start, buf, event_len); if (flags2 & FL_ORIG_EVENT_SIZE) { DBUG_ASSERT(seq_no == 1); original_event_size= uint8korr((post_start)); post_start+= 8; start_offset+= 8; VALIDATE_BYTES_READ(post_start, buf, event_len); DBUG_ASSERT(original_event_size); } /* post_start so far only has read header data, no data from the rows event itself. Ensure there is still data for the rows event. */ DBUG_ASSERT(static_cast<uint>(post_start - ev_buffer_base) < event_len); DBUG_VOID_RETURN; } bool Partial_rows_log_event::is_valid() const { bool is_first= seq_no == 1; /* There should be at least 2 fragments for the group, and the sequence number of this instance should fall between 1 and the total number of fragments. */ bool has_valid_sequencing= seq_no >= 1 && total_fragments >= 2 && seq_no <= total_fragments; bool has_valid_orig_event_size= ((flags2 & FL_ORIG_EVENT_SIZE) && is_first && original_event_size) || !(flags2 & FL_ORIG_EVENT_SIZE); /* To be valid from a fragmentation context, the following should hold true: 1) There should be a _valid_ Rows_log_event that is referenced 2) Start_offset and end_offset should be representative of the fragment. If it is the first fragment, then start_offset should be 0. The end_offset should be after the start_offset. 3) If it is the first event, we should write metadata and the original event size; otherwise, neither should be written. 4) ev_buffer_base should not be used */ bool is_valid_for_fragmentation= (rows_event && rows_event->is_valid()) && (((start_offset || is_first) && end_offset) && (end_offset > start_offset)) && ((is_first && metadata_written) || !metadata_written) && !ev_buffer_base; /* To be valid from an assembly context, the following should hold true: 1) ev_buffer_base should be set (referencing the read-in buffer) 2) start_offset should point beyond the Partial_rows_log_event header, as it points to the start of the Rows_log_event content 3) end_offset should be non-zero (there should always be some content) 4) Neither rows_event nor metadata_written should be set */ bool is_valid_for_assembly= ev_buffer_base && start_offset > PARTIAL_ROWS_HEADER_LEN && end_offset && (!rows_event && !metadata_written); bool is_valid= has_valid_sequencing && has_valid_orig_event_size && (is_valid_for_fragmentation ^ is_valid_for_assembly); return is_valid; } 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
¤ Die Informationen auf dieser Webseite wurden nach bestem Wissen sorgfältig zusammengestellt. Es wird jedoch weder Vollständigkeit, noch Richtigkeit, noch Qualität der bereit gestellten Informationen zugesichert.0.233Bemerkung: (vorverarbeitet am 2026-10-08) ¤*Bot Zugriff |
|
||||||||||||||
2026-10-09