diff --git a/sql/log.cc b/sql/log.cc index 5d6d6381577..9d3d7d4c751 100644 --- a/sql/log.cc +++ b/sql/log.cc @@ -8145,6 +8145,7 @@ Event_log::flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event, { Log_event_writer writer(&cache_data->cache_log, cache_data, pending->select_checksum_alg(cache_data), NULL); + Log_event *ev_to_write= NULL; /* Write pending event to the cache. @@ -8160,14 +8161,45 @@ Event_log::flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event, clear_dbug= true; } }); - if (writer.write(pending)) + + ulong rows_ev_metadata_len= + LOG_EVENT_HEADER_LEN + ROWS_HEADER_LEN_V1 + BINLOG_CHECKSUM_LEN; + ulong max_rows_ev_len= + opt_binlog_row_event_fragment_threshold - rows_ev_metadata_len; + if (pending->rows_data_size_exceeds(max_rows_ev_len)) + { + Rows_log_event_fragmenter fragmenter= Rows_log_event_fragmenter( + thd, is_transactional, opt_binlog_row_event_fragment_threshold, + pending); + Rows_log_event_fragmenter::Fragmented_rows_log_event *frag_ev; + if (!(frag_ev= fragmenter.fragment())) + { + delete pending; + cache_data->set_pending(NULL); + DBUG_RETURN(1); + + } + DBUG_ASSERT(frag_ev->is_valid()); + ev_to_write= frag_ev; + } + else + { + ev_to_write= pending; + } + + DBUG_ASSERT(ev_to_write); + if (writer.write(ev_to_write)) { set_write_error(thd, is_transactional); if (check_cache_error(thd, cache_data) && (stmt_has_updated_non_trans_table(thd) || !is_transactional)) cache_data->set_incident(); + + if (ev_to_write && ev_to_write != pending) + delete ev_to_write; delete pending; + cache_data->set_pending(NULL); DBUG_EXECUTE_IF("simulate_disk_full_at_flush_pending", { @@ -8182,6 +8214,8 @@ Event_log::flush_and_set_pending_rows_event(THD *thd, Rows_log_event* event, DBUG_SET("-d,simulate_file_write_error"); }); + if (ev_to_write && ev_to_write != pending) + delete ev_to_write; delete pending; } diff --git a/sql/log_event.cc b/sql/log_event.cc index c6d218c9b48..98aa5c4f6a6 100644 --- a/sql/log_event.cc +++ b/sql/log_event.cc @@ -697,6 +697,7 @@ const char* Log_event::get_type_str(Log_event_type type) 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 */ } @@ -1267,6 +1268,10 @@ Log_event *Log_event::read_log_event_no_checksum( ev= new Table_map_log_event(buf, static_cast(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(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(event_len), @@ -2272,6 +2277,7 @@ Format_description_log_event(uint8 binlog_ver, const char* server_ver, 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; @@ -2479,7 +2485,7 @@ Format_description_log_event::is_version_before_checksum(const master_version_sp checksum-unaware (effectively no checksum) and the actual [1-254] range alg descriptor. */ -enum_binlog_checksum_alg get_checksum_alg(const uchar *buf, ulong len) +enum_binlog_checksum_alg get_checksum_alg(const uchar *buf, size_t len) { enum_binlog_checksum_alg ret; char version[ST_SERVER_VER_LEN]; @@ -4070,6 +4076,111 @@ Incident_log_event::Incident_log_event(const uchar *buf, uint event_len, } +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(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) diff --git a/sql/log_event.h b/sql/log_event.h index 32c0e5accd7..3f8b0e47a06 100644 --- a/sql/log_event.h +++ b/sql/log_event.h @@ -225,6 +225,7 @@ class String; #define GTID_LIST_HEADER_LEN 4 #define START_ENCRYPTION_HEADER_LEN 0 #define XA_PREPARE_HEADER_LEN 0 +#define PARTIAL_ROWS_HEADER_LEN (4 + 4 + 1) /* Max number of possible extra bytes in a replication event compared to a @@ -406,6 +407,12 @@ class String; #define ELQ_FN_POS_END_OFFSET ELQ_FILE_ID_OFFSET + 8 #define ELQ_DUP_HANDLING_OFFSET ELQ_FILE_ID_OFFSET + 12 +/* PRW = "Partial RoWs" */ +#define PRW_TOTAL_SEQS_OFFSET 0 +#define PRW_SELF_SEQ_OFFSET 4 +#define PRW_FLAGS_OFFSET 8 +#define PRW_OPTIONAL_DATA_OFFSET 9 + /* 4 bytes which all binlogs should begin with */ #define BINLOG_MAGIC (const uchar*) "\xfe\x62\x69\x6e" @@ -747,6 +754,11 @@ enum Log_event_type UPDATE_ROWS_COMPRESSED_EVENT = 170, DELETE_ROWS_COMPRESSED_EVENT = 171, + /* + Partial Row Data Event + */ + PARTIAL_ROW_DATA_EVENT = 172, + /* Add new MariaDB events here - right above this comment! */ ENUM_END_EVENT /* end marker */ @@ -1049,13 +1061,19 @@ public: checksum_len(( checksum_alg != BINLOG_CHECKSUM_ALG_OFF && checksum_alg != BINLOG_CHECKSUM_ALG_UNDEF) ? BINLOG_CHECKSUM_LEN : 0), - file(file_arg), cache_data(cache_data_arg), crypto(cr) { } + file(file_arg), cache_data(cache_data_arg), crc(0), crypto(cr) { } private: IO_CACHE *file; binlog_cache_data *cache_data; /** Placeholder for event checksum while writing to binlog. + + The crc is reset after writing a binary log event's footer (which is where + the crc is written). It isn't reset when starting a new event (i.e. when + writing the header) because Partial_rows_log_events will write the header + for both its own Partial_rows_log_event, as well as the header for the + Rows_log_event which is being fragmented. */ ha_checksum crc; /** @@ -4728,6 +4746,17 @@ public: virtual ~Rows_log_event(); + /* + Check if the length of the rows data exceeds a given limit. This is used to + decide whether or not to fragment the event into multiple partial row log + events when binlogging. + */ + my_bool rows_data_size_exceeds(ulong size_limit) + { + ulong const data_size= static_cast(m_rows_cur - m_rows_buf); + return data_size > size_limit; + } + void set_flags(flag_set flags_arg) { m_flags |= flags_arg; } void clear_flags(flag_set flags_arg) { m_flags &= ~flags_arg; } flag_set get_flags(flag_set flags_arg) const { return m_flags & flags_arg; } @@ -5098,6 +5127,8 @@ private: */ virtual int do_exec_row(rpl_group_info *rli) = 0; #endif /* defined(MYSQL_SERVER) && defined(HAVE_REPLICATION) */ + + friend class Rows_log_event_fragmenter; }; /** @@ -5545,6 +5576,284 @@ bool copy_cache_to_file_wrapped(IO_CACHE *body, bool is_verbose); #endif +/** + @class Partial_rows_log_event + + When any given instantiation of a Rows_log_event (e.g. Write_rows_log_event, + etc) is too large to be sent to a replica (i.e. larger than the value + slave_max_allowed_packet, as configured on a replica), then the rows event + must be fragmented into sub-events (i.e. Partial_rows_log_events) so the + event can be transmitted to the replica. The size of each event is configured + via the system variable binlog_row_event_fragment_threshold. The replica will + then take the content of each of these Partial_rows_log_events, and join them + together into a large Rows_log_event to be executed as normal. + + Partial_rows_log_events are written to the binary log sequentially, and + the replica assembles the events in the order they are binlogged. Each + Partial_rows_log_event stores its sequence number (seq_no) in the overall + series of fragments, the total number of fragments needed to re-assemble the + Rows_log_event (total_fragments), a uchar for flags for embedding extra data, + and any additional data as specified by the flags. Currently, only the first + event in a grouping will have additional data: it will set the first bit in + the flags (FL_ORIG_EVENT_SIZE) to indicate it will be storing the total size + of the underlying Rows_log_event. + + The cached Rows_log_event data is fragmented into Partial_rows_log_events as + follows. The primary will still generate a Rows_log_event to write to the + binlog; however, during the actual writing process, the raw data of the + rows event is split into fragments, each covering some continuous section of + the rows data. A Partial_rows_log_event is created for each continuous + section, and the Partial_rows_log_events are written sequentially in-place of + the too-large Rows_log_event. The original data to be fragmented will include + a header and data header; however, will not include a checksum, as each + Partial_rows_log_event will have a checksum for validation, as well as a + sequence_number and total number of fragments to ensure all fragments are + present. + + @note This class behaves slightly different depending on if it is being used + for fragmentation (writing) or assembly (reading). During fragmentation, each + Partial_rows_log_event uses its member variables start_offset and end_offset + to refer to the boundaries of the original Rows_log_event that this + particular class is responsible for writing (specifically, it references only + the rows data; it is implied that the first event of the group should also + write the header and data header). During assembly, the start_offset and + end_offset refer to the start and end of the rows_event data within the + context of the full Partial_rows_log_event data. + + @note Where slave_max_allowed_packet has a maximum of 1GB at the time of + creating this class (MariaDB 12.3), even if that one day grows larger, the + current binlog format only supports row events of maximum size 4GB anyway. + If/when this max_allowed_packet restriction is ever lifted, the size of any + given Partial_rows_log_event should then be 4GB. + + @section Partial_rows_log_event_binary_format Binary format + + See @ref Log_event_binary_format "Binary format for log events" for + a general discussion and introduction to the binary format of binlog + events. + + The Post-Header has five components: + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
Post-Header for Partial_rows_log_event
NameFormatDescription
Sequence Number4 byte unsigned integer + Where the content of a group of Partial_rows_log_events is put + together in-order to recreate an original Rows_log_event, the sequence + number specifies the position of a single Partial_rows_log_event in + relation to the other events in the group. +
Total Fragments4 byte unsigned integer + The total number of Partial_rows_log_events that exist to recreate the + original Rows_log_event +
Flags1 byte for flags + Flags to specify optional pieces of data for the event. Current flags + are: + (1) FL_ORIG_EVENT_SIZE - Used to provide the total size of the + underlying Rows_log_event (i.e. the full size, when re-assembled + on the slave). +
Reserved space (see flags bit 1)0 bytes / 8 bytes + Reserved bytes for storing optional data when indicated by flags +
+ + The Body has the following components: + + + + + + + + + + + + + + + +
Body for Partial_rows_log_event
NameFormatDescription
contentString + A fragment of the original Rows_log_event content. +
+*/ +class Partial_rows_log_event : public Log_event +{ +public: + /* + Flags for future extensibility, currently unused + */ + uchar flags2; + + /* + Within the context of a fragmented Rows_log_event, the sequence number for + _this_ particular fragment. Starts at 1. + */ + uint32_t seq_no; + + /* + The total number of fragments to store the content of a Rows_log_event that + is written in fragments. + */ + uint32_t total_fragments; + + /* + The size of the original Rows_log_event. + */ + uint64_t original_event_size; + + /* + The size of the header and data_header that is written for the first + Partial_rows_log_event in a group. This is needed to calculate the + start_offset and end_offset to still fit within the desired size of a + fragment. That is, because events are written as (pseudo-code) + + if (is_first_in_group(partial_ev)) + { + Log_event_writer::write_header() + Log_event_writer::write_data_header() + Log_event_writer::write_data_body_metadata() + } + Log_event_writer::write_data_body_rows(start_offset, end_offset) + + for the first Partial_rows_log_event written, we need to track the size + of data written by the header, data_header, and data_body_metadata to + make sure that the start_offset and end_offset account for this to not + write too large of an event. + */ + uint32_t metadata_written; + + /* + Functions differently depending if fragmenting (writing to binlog) or + assembling (reading from binlog). + + During fragmentation, this refers to the offset in the rows_data of the + Rows_log_event where this object is responsible for writing. + + During assembly, this refers to the offset in the memory buffer of the + Partial_rows_log_event where the Rows_log_event content starts. + */ + uint64_t start_offset; + + /* + Functions differently depending if fragmenting (writing to binlog) or + assembling (reading from binlog). + + During fragmentation, this refers to the end offset in the rows_data of the + Rows_log_event where this object is responsible for writing. + + During assembly, this is the _length_ of the Rows_log_event data in the + scope of the overall Partial_rows_log_event data buffer. + */ + uint64_t end_offset; + + /* + During assembly, this refers to the start of the memory buffer of the + overall Partial_rows_log_event. + + Unused for fragmentation. + */ + const uchar *ev_buffer_base; + + /* FL_ORIG_EVENT_SIZE is set when the original event size is written */ + static const uchar FL_ORIG_EVENT_SIZE= 1; + +private: + /* + During fragmentation, references the original Rows_log_event that is to + be fragmented. + + Unused for assembly. + */ + Rows_log_event *rows_event; + + + static const uint max_data_length= + PARTIAL_ROW_DATA_EVENT + + sizeof(decltype(Partial_rows_log_event::original_event_size)); + +public: +#ifdef MYSQL_SERVER + Partial_rows_log_event(THD *thd, bool is_transactional, uint32 seq_no, + uint32 total_fragments, uint64_t original_event_size, + uint32 metadata_written, uint64_t start_offset, + uint64_t end_offset, uchar flags2, + Rows_log_event *rows_event) + : Log_event(thd, 0, is_transactional), flags2(flags2), seq_no(seq_no), + total_fragments(total_fragments), + original_event_size(original_event_size), + metadata_written(metadata_written), start_offset(start_offset), + end_offset(end_offset), ev_buffer_base(NULL), rows_event(rows_event){}; +#endif /* MYSQL_SERVER */ + + Partial_rows_log_event( + const uchar *buf, uint event_len, + const Format_description_log_event *description_event); + + ~Partial_rows_log_event() {} + + /* + For the event to be valid, it must hold true to some general principles for + both fragmentation and assembly; though be set up specifically for the + fragmentation or assembly use case (and not both). + */ + bool is_valid() const override; + + Log_event_type get_type_code() override { + return PARTIAL_ROW_DATA_EVENT; + } + + int get_data_size() override + { + return (int) get_rows_size() + metadata_written + PARTIAL_ROWS_HEADER_LEN + + (flags2 & FL_ORIG_EVENT_SIZE ? 8 : 0); + } + + size_t get_rows_size() + { + return end_offset - start_offset; + } + +#ifdef MYSQL_SERVER + bool write_data_header(Log_event_writer *writer) override; + bool write_data_body(Log_event_writer *writer) override; +#ifdef HAVE_REPLICATION + int do_apply_event(rpl_group_info *rgi) override; + void pack_info(Protocol* protocol) override; +#endif /* HAVE_REPLICATION */ +#else + bool print(FILE* file, PRINT_EVENT_INFO* print_event_info) override; +#endif +}; + + #ifdef MYSQL_SERVER /***************************************************************************** @@ -5588,6 +5897,226 @@ inline int Log_event_writer::write(Log_event *ev) return res; } +#ifdef MYSQL_SERVER +/* + Helper class which fragments the content of a large Rows_log_event into + multiple Partial_rows_log_events, which can be written to the binary log. + Writing these Partial_rows_log_events is abstracted by a special Log_event + implementation, a Fragmented_rows_log_event, which contains and writes a + complete group of multiple Partial_rows_log_events (i.e. all these partial + events can be assembled to recreate the original Rows_log_event). That way, + it is easy to pass a single Fragmented_rows_log_event to + Log_event_writer::write(), and it will write all Partial_rows_log_events to + the binary log. +*/ +class Rows_log_event_fragmenter +{ +public: + /* + Helper class for writing a complete group of Partial_row_log_events that + store the full content of a large Rows_log_event. It really only + implements Log_event::write(), such that it will write all + Partial_rows_log_events into the binary log. It also implements + Log_event::is_valid() to ensure the fragmentation was done correctly. + */ + class Fragmented_rows_log_event : public Log_event + { + private: + Partial_rows_log_event *fragments; + uint32_t n_fragments; + public: + Fragmented_rows_log_event(Partial_rows_log_event *fragments, + uint32_t n_fragments) + : fragments(fragments), n_fragments(n_fragments){}; + + ~Fragmented_rows_log_event() + { + DBUG_ASSERT(n_fragments); + my_free(fragments); + } + + /* + Write all Partial_rows_log_event fragments + */ + bool write(Log_event_writer *writer) override; + + /* + Valid if all fragments are correctly ordered, and each fragment itself + is valid + */ + bool is_valid() const override; + + /* Not applicable */ + Log_event_type get_type_code() override { + DBUG_ASSERT(0); + return UNKNOWN_EVENT; + } + + /* Not applicable */ + int get_data_size() override + { + DBUG_ASSERT(0); + return 0; + } + +#ifdef MYSQL_SERVER + /* Not applicable */ + bool write_data_header(Log_event_writer *writer) override + { + DBUG_ASSERT(0); + return 1; + } + + /* Not applicable */ + bool write_data_body(Log_event_writer *writer) override + { + DBUG_ASSERT(0); + return 1; + } +#endif + }; + +public: + + THD *thd; + bool is_transactional; + + /* + Maximum size of a Partial_rows_log_event + */ + uint32_t fragment_size; + + /* + The Rows_log_event to fragment + */ + Rows_log_event *rows_event; + + /* + {{num_chunks}} Partial_rows_log_events that altogether hold the entire + content of a Rows_log_event. + */ + Partial_rows_log_event *fragments; + +public: + Rows_log_event_fragmenter(THD *thd, bool is_transactional, + uint32_t fragment_size, Rows_log_event *rows_event) + : thd(thd), is_transactional(is_transactional), + fragment_size(fragment_size), rows_event(rows_event) + { + } + + /* + Note that our variable fragments which we allocate is freed by + Fragmented_rows_log_event::~Fragmented_rows_log_event() + */ + ~Rows_log_event_fragmenter() {} + + /* + Fragments our Rows_log_event (provided in the constructor) into num_chunks + Partial_rows_log_events, housed by a Fragmented_rows_log_event. + */ + Fragmented_rows_log_event* fragment(); + + /* + The amount of Rows_log_event content to be held by each + Partial_rows_log_event. Note that this doesn't consider the metadata + of the first event group, i.e. the metadata for the rows log event and + additional data indicated by flags2. That is manually accounted for when + organizing/computing the fragments. + */ + uint32_t get_payload_size_per_chunk() + { + DBUG_ASSERT(fragment_size > LOG_EVENT_HEADER_LEN + + PARTIAL_ROWS_HEADER_LEN + + BINLOG_CHECKSUM_LEN); + return fragment_size - LOG_EVENT_HEADER_LEN - PARTIAL_ROWS_HEADER_LEN - + BINLOG_CHECKSUM_LEN; + } + + Partial_rows_log_event *get_fragment(size_t i) + { + return &fragments[i]; + } +}; +#endif /* MYSQL_SERVER */ + +/* + Class which assembles a Rows_log_event from a group of + Partial_rows_log_events +*/ +class Rows_log_event_assembler +{ +public: + /* + State tracker to ensure that Partial_rows_log_events are being ingested + in-order + */ + uint32_t last_fragment_seen; + + + /* + Memory buffer, initialized to the maximum size of the Rows_log_event + (i.e. total_fragments * event_size), where each Partial_rows_log_event + will append its Rows_log_event content to. + */ + char *rows_ev_buf_builder_ptr; + + rpl_group_info *rgi; + + /* + Total length of the Rows_log_event to construct + */ + uint64_t ev_len; + + /* + Total number of Partial_rows_log_events in the group. + */ + uint32_t total_fragments; + + /* + Used to track whether or not we need to free rows_ev_buf_builder_ptr + during destruction. If the Rows_log_event is successfully created, + the allocated rows_ev_buf_builder_ptr is transferred to the created + Rows_log_event for its memory management. Otherwise, e.g if there is an + error that prevents creation of the event, we will need to delete that + memory ourselves. + */ + my_bool row_ev_created; + + Rows_log_event_assembler(rpl_group_info *rgi, uint32_t total_fragments) + : last_fragment_seen(0), rows_ev_buf_builder_ptr(NULL), rgi(rgi), + total_fragments(total_fragments), row_ev_created(false){}; + + ~Rows_log_event_assembler() + { + /* + The rows_ev_buf_builder_ptr is supposed to be freed by the Rows_log_event + which we create; but if this object is destroyed before the + Rows_log_event is created (e.g. if the slave errors), we need to free the + buffer ourselves + */ + if (!row_ev_created && rows_ev_buf_builder_ptr) + my_free(rows_ev_buf_builder_ptr); + }; + + bool all_fragments_assembled() + { + return last_fragment_seen == total_fragments; + } + + /* + Append the Rows_log_event content from a Partial_rows_log_event into our + builder to re-create the original Rows_log_event. + */ + int append(Partial_rows_log_event *partial_ev); + + /* + Build the original Rows_log_event once all Partial_rows_log_event + fragments have been appended. + */ + Log_event *create_rows_event(const Format_description_log_event *fdle); +}; + /** The function is called by slave applier in case there are active table filtering rules to force gathering events associated @@ -5600,7 +6129,7 @@ bool slave_execute_deferred_events(THD *thd); bool event_that_should_be_ignored(const uchar *buf); bool event_checksum_test(uchar *buf, size_t event_len, enum_binlog_checksum_alg alg); -enum_binlog_checksum_alg get_checksum_alg(const uchar *buf, ulong len); +enum_binlog_checksum_alg get_checksum_alg(const uchar *buf, size_t len); extern TYPELIB binlog_checksum_typelib; #ifdef WITH_WSREP enum Log_event_type wsrep_peak_event(rpl_group_info *rgi, ulonglong* event_size); diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc index acbc36abf54..00a5e1de525 100644 --- a/sql/log_event_server.cc +++ b/sql/log_event_server.cc @@ -804,7 +804,7 @@ int Log_event_writer::write_header(uchar *pos, size_t len) { uchar save=pos[FLAGS_OFFSET]; pos[FLAGS_OFFSET]&= ~LOG_EVENT_BINLOG_IN_USE_F; - crc= my_checksum(0, pos, len); + crc= my_checksum(crc, pos, len); pos[FLAGS_OFFSET]= save; } @@ -848,17 +848,28 @@ int Log_event_writer::write_footer() uchar checksum_buf[BINLOG_CHECKSUM_LEN]; int4store(checksum_buf, crc); if ((this->*encrypt_or_write)(checksum_buf, BINLOG_CHECKSUM_LEN)) + { + crc= 0; DBUG_RETURN(ER_ERROR_ON_WRITE); + } } if (ctx) { uint dstlen; uchar dst[MY_AES_BLOCK_SIZE*2]; if (encryption_ctx_finish(ctx, dst, &dstlen)) + { + crc= 0; DBUG_RETURN(1); + } if (maybe_write_event_len(dst, dstlen) || write_internal(dst, dstlen)) + { + crc= 0; DBUG_RETURN(ER_ERROR_ON_WRITE); + } } + + crc= 0; DBUG_RETURN(0); } @@ -4853,17 +4864,6 @@ int Rows_log_event::do_add_row_data(uchar *row_data, size_t length) DBUG_EXECUTE_IF("simulate_too_big_row_case4", cur_size= UINT_MAX32 - (block_size * 10); length= (block_size * 10) - block_size + 1;); - size_t remaining_space= UINT_MAX32 - cur_size; - /* Check that the new data fits within remaining space and we can add - block_size without wrapping. - */ - if (cur_size > UINT_MAX32 || length > remaining_space || - ((length + block_size) > remaining_space)) - { - sql_print_error("The row data is greater than 4GB, which is too big to " - "write to the binary log."); - DBUG_RETURN(ER_BINLOG_ROW_LOGGING_FAILED); - } size_t const new_alloc= block_size * ((cur_size + length + block_size - 1) / block_size); @@ -5714,6 +5714,448 @@ bool Rows_log_event::write_data_body(Log_event_writer *writer) } +/************************************************************************** + Partial_rows_log_event member functions +**************************************************************************/ + +#ifdef HAVE_REPLICATION +int Partial_rows_log_event::do_apply_event(rpl_group_info *rgi) +{ + int res= 0; + DBUG_ASSERT(rgi); + DBUG_ASSERT(rgi->thd); + Rows_log_event_assembler *assembler; + PSI_stage_info org_stage; + rgi->thd->backup_stage(&org_stage); + + if (!rgi->assembler) + { + rgi->assembler= (Rows_log_event_assembler *) my_malloc( + PSI_INSTRUMENT_ME, sizeof(Rows_log_event_assembler), MYF(MY_WME)); + if (!rgi->assembler) + { + rgi->rli->report(ERROR_LEVEL, my_errno, rgi->gtid_info(), + "Could not allocate Rows_log_event_assembler"); + res= ER_OUTOFMEMORY; + goto end; + } + new (rgi->assembler) Rows_log_event_assembler(rgi, total_fragments); + } + assembler= rgi->assembler; + + THD_STAGE_INFO(rgi->thd, stage_buffer_partial_rows); + if ((res= assembler->append(this))) + { + rgi->assembler->~Rows_log_event_assembler(); + my_free(rgi->assembler); + rgi->assembler= NULL; + goto end; + } + + if (assembler->all_fragments_assembled()) + { + THD_STAGE_INFO(rgi->thd, stage_constructing_rows_ev); + Log_event *ev= assembler->create_rows_event( + rgi->rli->relay_log.description_event_for_exec); + rgi->assembler->~Rows_log_event_assembler(); + my_free(rgi->assembler); + rgi->assembler= NULL; + + if (!ev) + { + rgi->rli->report( + ERROR_LEVEL, ER_SLAVE_RELAY_LOG_READ_FAILURE, NULL, + ER_THD(thd, ER_SLAVE_RELAY_LOG_READ_FAILURE), + "Could not parse Rows_log_event re-assembled from " + "Partial_rows_log_events. The possible reasons are: the master's " + "binary log is corrupted (you can check this by running " + "'mysqlbinlog' on the binary log), the slave's relay log is " + "corrupted (you can check this by running 'mysqlbinlog' on the " + "relay log), a network problem, or a bug in the master's or slave's " + "MariaDB code. If you want to check the master's binary log or " + "slave's relay log, you will be able to know their names by issuing " + "'SHOW SLAVE STATUS' on this slave."); + res= 1; + goto end; + } + + /* + We can safely ignore the return value of + apply_event_and_update_pos_setup(), which indicates if this event should + be skipped or not. Consider the two cases in which this is called: + + 1) This event is being applied by the SQL thread directly. In which + case, in apply_event_and_update_pos(), the ev->shall_skip() check is + performed on the encompassing Partial_rows_log_event(), which will + duplicate the value/behavior of this assembled Rows_log_event but at + an earlier time. In other words, if the Partial_rows_log_event is + skipped, execution could not be here. + + 2) This event is applied by a SQL BINLOG base-64 event, in which case, + the skip-logic is not performed and we don't care about the result. + */ + apply_event_and_update_pos_setup(ev, thd, rgi); + + res= ev->apply_event(rgi); + delete ev; + } + +end: + THD_STAGE_INFO(rgi->thd, org_stage); + return res; +} +#endif + +bool Partial_rows_log_event::write_data_body(Log_event_writer *writer) +{ + uint64_t cur_offset= start_offset; + uint64_t row_data_len_to_write= end_offset - start_offset; + /* + Write the width and cols bitmap for the first event. This shouldn't ever + extend beyond one fragment, so don't add checks to split these. + */ + my_bool first_fragment= !cur_offset; + if (first_fragment) + { + rows_event->write_header(writer, rows_event->get_data_size()); + rows_event->write_data_header(writer); + rows_event->write_data_body_metadata(writer); + } + + return rows_event->write_data_body_rows(writer, start_offset, + row_data_len_to_write); +} + +bool Partial_rows_log_event::write_data_header(Log_event_writer *writer) +{ + uchar buf[max_data_length]; // No need to init the buffer + + /* + The length of the amount of data that will be written. This is also used + to track where to write optional fields. + */ + size_t header_size= PARTIAL_ROWS_HEADER_LEN; + + /* + Mandatory fields occuring in all Partial_rows_log_events + */ + int4store(buf + PRW_TOTAL_SEQS_OFFSET, this->total_fragments); + int4store(buf + PRW_SELF_SEQ_OFFSET, this->seq_no); + buf[PRW_FLAGS_OFFSET]= this->flags2; + + /* + Optional fields that may be written depending on flags2 + */ + if(flags2 & FL_ORIG_EVENT_SIZE) + { + DBUG_ASSERT(original_event_size && seq_no == 1); + int8store(buf + header_size, original_event_size); + header_size+= 8; + } + + return write_data(writer, buf, header_size); +} + +#if defined(HAVE_REPLICATION) +void Partial_rows_log_event::pack_info(Protocol *protocol) +{ + char buf[256]; + size_t bytes= my_snprintf(buf, sizeof(buf), + "Fragment %u of %u", seq_no, total_fragments); + protocol->store(buf, bytes, &my_charset_bin); +} +#endif + +bool Rows_log_event_fragmenter::Fragmented_rows_log_event::write( + Log_event_writer *writer) +{ + for (uint32_t i= 0; i < n_fragments; i++) + { +#ifndef DBUG_OFF + bool skip_writing_pev= + (DBUG_IF("partial_rows_skip_binlogging_first_fragment") && i == 0) || + (DBUG_IF("partial_rows_skip_binlogging_middle_fragment") && i == 1) || + (DBUG_IF("partial_rows_skip_binlogging_last_fragment") && + i == n_fragments - 1); +#endif + + bool res= +#ifndef DBUG_OFF + !skip_writing_pev && +#endif + writer->write(&fragments[i]); + + if (res) + return res; + } + return 0; +} + +bool Rows_log_event_fragmenter::Fragmented_rows_log_event::is_valid() const +{ + uint32_t last_fragment_seen= 0; + for (uint32_t i= 0; i < n_fragments; i++) + { + Partial_rows_log_event *frag= &fragments[i]; + bool is_valid= (frag->total_fragments == n_fragments) && + (frag->seq_no == last_fragment_seen + 1) && + frag->is_valid(); + if (!is_valid) + return false; + last_fragment_seen= frag->seq_no; + } + return true; +} + +/* + Fragments a Rows_log_event into multiple Partial_rows_log_event fragments. + It is assumed that the Rows_log_event_fragmenter already has the source + Rows_log_event at this point. To fragment into a group of + Partial_rows_log_event, this function first allocates a chunk of memory to + hold all fragmented events. To calculate the size of memory required, the + total size of the Rows_log_event is divided by the amount of data that each + Partial_rows_log_event can hold. The first fragment also holds the original + size of the Rows event, and for the calculation, this size is aggregated into + the Rows_log_event total size, as it is only applicable to one event in the + group. + + The group of Partial_rows_log_events therefore looks like: + + Fragment 1: + 1. Common header for the Partial_rows_log_event + 2. Post-header for the Partial_rows_log_event + * Total number of fragments + * Sequence number of this event + * Original size of the Rows_log_event + 3. Rows log event data + * Common header for the Rows_log_event + * Post-header for the Rows_log_event + * Metadata for the Rows_log_event (i.e. the width and columns bitmap) + * Rows data up to the end of the fragment (excluding the checksum) + 4. Checksum for the Partial_rows_log_event + + Fragment 2 through (n-1): + 1. Common header for the Partial_rows_log_event + 2. Post-header for the Partial_rows_log_event + * Total number of fragments + * Sequence number of this event + 3. Rows log event data + * Rows data up to the end of the fragment (excluding the checksum) + 4. Checksum for the Partial_rows_log_event + + Fragment n (last fragment): + 1. Common header for the Partial_rows_log_event + 2. Post-header for the Partial_rows_log_event + * Total number of fragments + * Sequence number of this event + 3. Rows log event data + * Remaining rows data + 4. Checksum for the Partial_rows_log_event +*/ +Rows_log_event_fragmenter::Fragmented_rows_log_event * +Rows_log_event_fragmenter::fragment() +{ + Fragmented_rows_log_event *ev; + uchar width_tmp_buf[MAX_INT_WIDTH]; + uchar *const width_tmp_buf_end= + net_store_length(width_tmp_buf, (size_t) rows_event->m_width); + uint32_t width_size= + static_cast(width_tmp_buf_end - width_tmp_buf); + + /* + Update row events write an extra bitmap + */ + uint32_t cols_size= + no_bytes_in_export_map(&rows_event->m_cols) * + ((rows_event->get_general_type_code() == UPDATE_ROWS_EVENT) ? 2 : 1); + + + /********************************************************************** + Attributes about the length of the underlying Rows_log_event + **********************************************************************/ + /* + The size of the Rows_log_event header and metadata (table width, column + bitmap) + */ + uint32_t rows_ev_metadata_size= + LOG_EVENT_HEADER_LEN + ROWS_HEADER_LEN_V1 + width_size + cols_size; + + /* The size of the actual row data payload */ + uint64_t rows_ev_data_size= + static_cast(rows_event->m_rows_cur - rows_event->m_rows_buf); + + /* + The total size of the original event that will be re-created on the slave + */ + uint64_t rows_ev_total_size= + static_cast(rows_ev_metadata_size) + rows_ev_data_size; + /*********************************************************************/ + + + /********************************************************************** + Attributes to describe the encompassing Partial_rows_log_event group + **********************************************************************/ + /* + Extra payload in the first fragment: the original Rows_log_event metadata + plus the 8-byte original event size field. + */ + uint32_t first_ev_extra_size= rows_ev_metadata_size + 8 /* orig_event_size */; + + /* + The total data stream to be fragmented, including the extra data for the + first fragment. The extra data is included into this variable because it + simplifies the calculation, as it is only added once. + */ + uint64_t group_total_size= + static_cast(first_ev_extra_size) + rows_ev_data_size; + /*********************************************************************/ + + /* The maximum amount of payload data each fragment can hold. */ + uint32_t data_size_per_chunk= get_payload_size_per_chunk(); + + uint32_t last_chunk_size= (group_total_size % data_size_per_chunk); + uint8_t last_chunk= last_chunk_size ? 1 : 0; + uint32_t num_chunks= + static_cast((group_total_size / data_size_per_chunk)) + + last_chunk; + + fragments= + DBUG_IF("oom_fragmenting_large_rows_ev") + ? NULL + : (Partial_rows_log_event *) my_malloc( + PSI_INSTRUMENT_ME, sizeof(Partial_rows_log_event) * num_chunks, + MYF(MY_WME)); + DBUG_EXECUTE("oom_fragmenting_large_rows_ev", my_errno= ENOMEM;); + + if (!fragments) + { + my_error(ER_OUTOFMEMORY, MYF(0), sizeof(Partial_rows_log_event)*num_chunks); + return NULL; + } + + + /* + Offset into the Rows_log_event data to start writing at, inclusive + */ + uint64_t chunk_start; + + /* + Offset into the Rows_log_event data to stop writing at, inclusive (?) + */ + uint64_t chunk_end; + + /* + First chunk + */ + { + chunk_start= 0; + chunk_end= data_size_per_chunk - first_ev_extra_size; + new (&fragments[0]) Partial_rows_log_event( + thd, is_transactional, 1, num_chunks, rows_ev_total_size, + rows_ev_metadata_size, chunk_start, chunk_end, + Partial_rows_log_event::FL_ORIG_EVENT_SIZE, rows_event); + } + + /* + The rest of the chunks + */ + for (uint32 chunk_idx= 1; chunk_idx < num_chunks; chunk_idx++) + { + my_bool is_last_chunk= (chunk_idx == (num_chunks - 1)); + chunk_start= ((chunk_idx * data_size_per_chunk) - first_ev_extra_size); + chunk_end= chunk_start + + (is_last_chunk ? (last_chunk_size) : (data_size_per_chunk)); + new (&fragments[chunk_idx]) Partial_rows_log_event( + thd, is_transactional, chunk_idx + 1, num_chunks, 0, 0, chunk_start, + chunk_end, 0, rows_event); + } + + ev= new Fragmented_rows_log_event(fragments, num_chunks); + return ev; +} + +#ifdef HAVE_REPLICATION +// Len of "%u / %u" +#define PARTIAL_ROWS_EVENT_BAD_STREAM_ERRSTR_LEN (10 + 3 + 10 + 1) +int Rows_log_event_assembler::append(Partial_rows_log_event *partial_ev) +{ + if ((partial_ev->total_fragments != this->total_fragments) || + (partial_ev->seq_no != this->last_fragment_seen + 1)) + { + char buf[PARTIAL_ROWS_EVENT_BAD_STREAM_ERRSTR_LEN]; + buf[PARTIAL_ROWS_EVENT_BAD_STREAM_ERRSTR_LEN - 1]= '\0'; + my_snprintf(buf, sizeof(buf), "%u / %u", partial_ev->seq_no, + partial_ev->total_fragments); + rgi->rli->report(ERROR_LEVEL, ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM, + rgi->gtid_info(), + ER_THD(rgi->thd, ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM), + buf, this->last_fragment_seen + 1, this->total_fragments); + return ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM; + } + + if (this->last_fragment_seen == 0) + { + DBUG_ASSERT(partial_ev->seq_no == 1 && + partial_ev->flags2 & + Partial_rows_log_event::FL_ORIG_EVENT_SIZE && + partial_ev->original_event_size); + rows_ev_buf_builder_ptr= + DBUG_IF("oom_reassembling_large_rows_ev_buf") + ? NULL + : (char *) my_malloc(PSI_INSTRUMENT_ME, + partial_ev->original_event_size, MYF(MY_WME)); + DBUG_EXECUTE("oom_reassembling_large_rows_ev_buf", my_errno= ENOMEM;); + ev_len= 0; + } + if (!rows_ev_buf_builder_ptr) + { + my_error(ER_OUTOFMEMORY, MYF(0), total_fragments*partial_ev->get_rows_size()); + rgi->rli->report( + ERROR_LEVEL, rgi->thd->get_stmt_da()->get_sql_errno(), + rgi->gtid_info(), + "Could not append Partial_rows_log_event %u / %u to internal " + "Rows_log_event buffer: %s", + partial_ev->seq_no, partial_ev->total_fragments, + rgi->thd->get_stmt_da()->message()); + return ER_OUTOFMEMORY; + } + + memcpy(rows_ev_buf_builder_ptr + ev_len, + partial_ev->ev_buffer_base + partial_ev->start_offset, + partial_ev->get_rows_size()); + ev_len+= partial_ev->get_rows_size(); + + last_fragment_seen= partial_ev->seq_no; + + return 0; +} +#endif + +Log_event *Rows_log_event_assembler::create_rows_event( + const Format_description_log_event *fdle) +{ + const char *error= NULL; + Log_event *res= NULL; + + if ((res= + DBUG_IF("fail_parsing_rows_ev_from_reassembly") + ? NULL + : Log_event::read_log_event_no_checksum( + (uchar *) rows_ev_buf_builder_ptr, ev_len, &error, fdle))) + { + row_ev_created= true; + res->register_temp_buf((uchar *) rows_ev_buf_builder_ptr, true); + } + else + { + DBUG_EXECUTE("fail_parsing_rows_ev_from_reassembly", error= "test error";); + sql_print_error("Error in Log_event::read_log_event(): '%s'", error); + } + + return res; +} + + bool Rows_log_event::write_compressed(Log_event_writer *writer) { uchar *m_rows_buf_tmp= m_rows_buf; diff --git a/sql/mysqld.cc b/sql/mysqld.cc index 59b2a7817bc..0d2eb41fc33 100644 --- a/sql/mysqld.cc +++ b/sql/mysqld.cc @@ -454,6 +454,7 @@ my_bool opt_noacl; my_bool sp_automatic_privileges= 1; ulong opt_binlog_rows_event_max_size; +ulong opt_binlog_row_event_fragment_threshold; ulong binlog_row_metadata; my_bool opt_binlog_gtid_index= TRUE; uint opt_binlog_gtid_index_page_size= 4096; @@ -9681,6 +9682,7 @@ PSI_stage_info stage_alter_inplace_prepare= { 0, "preparing for alter table", 0} PSI_stage_info stage_alter_inplace= { 0, "altering table", 0}; PSI_stage_info stage_alter_inplace_commit= { 0, "Committing alter table to storage engine", 0}; PSI_stage_info stage_apply_event= { 0, "Apply log event", 0}; +PSI_stage_info stage_buffer_partial_rows= { 0, "Buffering Partial_rows_log_event", 0}; PSI_stage_info stage_changing_master= { 0, "Changing master", 0}; PSI_stage_info stage_checking_master_version= { 0, "Checking master version", 0}; PSI_stage_info stage_checking_permissions= { 0, "checking permissions", 0}; @@ -9689,6 +9691,7 @@ PSI_stage_info stage_checking_query_cache_for_query= { 0, "Checking query cache PSI_stage_info stage_cleaning_up= { 0, "Reset for next command", 0}; PSI_stage_info stage_closing_tables= { 0, "closing tables", 0}; PSI_stage_info stage_connecting_to_master= { 0, "Connecting to master", 0}; +PSI_stage_info stage_constructing_rows_ev= { 0, "Constructing Rows_log_event from buffer of fragments", 0}; PSI_stage_info stage_converting_heap_to_myisam= { 0, "Converting HEAP to " TMP_ENGINE_NAME, 0}; PSI_stage_info stage_copying_to_group_table= { 0, "Copying to group table", 0}; PSI_stage_info stage_copying_to_tmp_table= { 0, "Copying to tmp table", 0}; @@ -9926,6 +9929,7 @@ PSI_stage_info *all_server_stages[]= & stage_binlog_processing_checkpoint_notify, & stage_binlog_stopping_background_thread, & stage_binlog_waiting_background_tasks, + & stage_buffer_partial_rows, & stage_changing_master, & stage_checking_master_version, & stage_checking_permissions, @@ -9936,6 +9940,7 @@ PSI_stage_info *all_server_stages[]= & stage_commit, & stage_commit_implicit, & stage_connecting_to_master, + & stage_constructing_rows_ev, & stage_converting_heap_to_myisam, & stage_copy_to_tmp_table, & stage_copying_to_group_table, diff --git a/sql/mysqld.h b/sql/mysqld.h index 5418f41efaa..2888932c560 100644 --- a/sql/mysqld.h +++ b/sql/mysqld.h @@ -247,6 +247,7 @@ extern ulonglong slave_max_statement_time; extern double slave_max_statement_time_double; extern double slave_abort_blocking_timeout; extern ulong opt_binlog_rows_event_max_size; +extern ulong opt_binlog_row_event_fragment_threshold; extern ulong binlog_row_metadata; extern my_bool opt_binlog_gtid_index; extern uint opt_binlog_gtid_index_page_size; @@ -520,6 +521,7 @@ extern PSI_stage_info stage_alter_inplace_prepare; extern PSI_stage_info stage_alter_inplace; extern PSI_stage_info stage_alter_inplace_commit; extern PSI_stage_info stage_after_apply_event; +extern PSI_stage_info stage_buffer_partial_rows; extern PSI_stage_info stage_changing_master; extern PSI_stage_info stage_checking_master_version; extern PSI_stage_info stage_checking_permissions; @@ -528,6 +530,7 @@ extern PSI_stage_info stage_checking_query_cache_for_query; extern PSI_stage_info stage_cleaning_up; extern PSI_stage_info stage_closing_tables; extern PSI_stage_info stage_connecting_to_master; +extern PSI_stage_info stage_constructing_rows_ev; extern PSI_stage_info stage_converting_heap_to_myisam; extern PSI_stage_info stage_copying_to_group_table; extern PSI_stage_info stage_copying_to_tmp_table; diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc index 2709988d263..3cb5ce95a6b 100644 --- a/sql/rpl_rli.cc +++ b/sql/rpl_rli.cc @@ -2174,7 +2174,7 @@ rpl_group_info::rpl_group_info(Relay_log_info *rli) gtid_ev_flags2(0), gtid_ev_flags_extra(0), gtid_ev_sa_seq_no(0), reserved_start_alter_thread(0), finish_event_group_called(0), rpt(NULL), start_alter_ev(NULL), direct_commit_alter(false), sa_info(NULL), - is_new_trans(false) + is_new_trans(false), assembler(NULL) { reinit(rli); bzero(¤t_gtid, sizeof(current_gtid)); @@ -2185,6 +2185,8 @@ rpl_group_info::rpl_group_info(Relay_log_info *rli) rpl_group_info::~rpl_group_info() { + DBUG_ASSERT(!assembler); + free_annotate_event(); delete deferred_events; mysql_mutex_destroy(&sleep_lock); diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h index 9b8ff5b9746..9898759c330 100644 --- a/sql/rpl_rli.h +++ b/sql/rpl_rli.h @@ -951,6 +951,16 @@ struct rpl_group_info bool direct_commit_alter; start_alter_info *sa_info; bool is_new_trans; // marker of start_new_trans context + + /* + Use the scope of an RGI to re-assemble partial row log events into the + original rows log event. That is, the re-assembly and execution of a row + event happens happens in Partial_rows_log_event::do_apply_event(), so then + regardless of if the slave is running on parallel or serial mode, the + logic remains the same. + */ + Rows_log_event_assembler *assembler; + rpl_group_info(Relay_log_info *rli_); ~rpl_group_info(); void reinit(Relay_log_info *rli); diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt index 9af0b5d427c..e570fa4d07c 100644 --- a/sql/share/errmsg-utf8.txt +++ b/sql/share/errmsg-utf8.txt @@ -12382,3 +12382,5 @@ ER_BINLOG_POS_INVALID eng "The binlog offset %llu is invalid" ER_READING_BINLOG_FILE eng "Error reading from page number %u in binlog file (error code: %d)" +ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM + eng "Broken Partial_rows_log_event stream. Found %s, expected Partial_rows_log_event %u / %u" diff --git a/sql/slave.cc b/sql/slave.cc index 3ed3825cafe..65d674186af 100644 --- a/sql/slave.cc +++ b/sql/slave.cc @@ -3590,7 +3590,7 @@ sql_delay_event(Log_event *ev, THD *thd, rpl_group_info *rgi) Split out so that it can run with rli->data_lock held in non-parallel replication, but without the mutex held in the parallel case. */ -static int +int apply_event_and_update_pos_setup(Log_event* ev, THD* thd, rpl_group_info *rgi) { DBUG_ENTER("apply_event_and_update_pos_setup"); @@ -3662,8 +3662,27 @@ apply_event_and_update_pos_apply(Log_event* ev, THD* thd, rpl_group_info *rgi, DBUG_SET_INITIAL("-d,inject_slave_sql_before_apply_event"); };); #endif + if (reason == Log_event::EVENT_SKIP_NOT) - exec_res= ev->apply_event(rgi); + { + if (rgi->assembler && ev->get_type_code() != PARTIAL_ROW_DATA_EVENT && + !ev->is_artificial_event() && !ev->is_relay_log_event()) + { + rgi->rli->report( + ERROR_LEVEL, ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM, rgi->gtid_info(), + ER_THD(rgi->thd, ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM), + ev->get_type_str(), rgi->assembler->last_fragment_seen + 1, + rgi->assembler->total_fragments); + exec_res= ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM; + + rgi->assembler->~Rows_log_event_assembler(); + my_free(rgi->assembler); + rgi->assembler= NULL; + } + + if (!exec_res) + exec_res= ev->apply_event(rgi); + } #ifdef WITH_WSREP if (WSREP(thd)) { diff --git a/sql/slave.h b/sql/slave.h index 0e57b467cb7..02555210f55 100644 --- a/sql/slave.h +++ b/sql/slave.h @@ -260,6 +260,8 @@ void set_slave_thread_default_charset(THD *thd, rpl_group_info *rgi); int rotate_relay_log(Master_info* mi); int has_temporary_error(THD *thd); int sql_delay_event(Log_event *ev, THD *thd, rpl_group_info *rgi); +int apply_event_and_update_pos_setup(Log_event *ev, THD *thd, + struct rpl_group_info *rgi); int apply_event_and_update_pos(Log_event* ev, THD* thd, struct rpl_group_info *rgi); int apply_event_and_update_pos_for_parallel(Log_event* ev, THD* thd, diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc index 8ca9e38f6f6..4a32880ab8d 100644 --- a/sql/sys_vars.cc +++ b/sql/sys_vars.cc @@ -6110,6 +6110,20 @@ static Sys_var_ulong Sys_opt_binlog_rows_event_max_size( VALID_RANGE(256, UINT_MAX32 - (UINT_MAX32 % 256)), DEFAULT(8192), BLOCK_SIZE(256)); +static Sys_var_ulong Sys_opt_binlog_partial_rows_event_max_size( + "binlog_row_event_fragment_threshold", + "When a Rows_log_event exceeds this threshold, it will be fragmented " + "into multiple Partial_rows_log_event events in the binary log, each of " + "this configured maximum size. That is, all Partial_rows_log_events up " + "to the last in the group will be this configured maximum size, and the " + "last event will take the remaining size. This is relevant for events " + "that would surpass slave_max_allowed_packet when sending to the slave, " + "and thereby a sensible value would reflect the slave's configured " + "slave_max_allowed_packet", + GLOBAL_VAR(opt_binlog_row_event_fragment_threshold), + CMD_LINE(REQUIRED_ARG), VALID_RANGE(1024, MAX_MAX_ALLOWED_PACKET), + DEFAULT(MAX_MAX_ALLOWED_PACKET), BLOCK_SIZE(1024)); + static Sys_var_on_access_global Sys_slave_net_timeout(