database/mariadb
1
0
Fork 0
mirror of https://github.com/MariaDB/server.git synced 2026-10-23 18:49:29 +03:00
Code Activity

MDEV-32570 (server): Fragment ROW replication events larger than slave_max_allowed_packet

Browse Source 
This patch solves two problems:
  1. Rows log events cannot be transmitted to the slave if their
     size exceeds slave_max_packet_size (max 1GB at the time of
     writing this patch, i.e. MariaDB 12.3)
  2. Rows log events cannot be binlogged if they are larger than
     4GB because the common binlog event header field event_len is
     32-bits.

This patch adds support for fragmenting large Rows_log_events
through a new event type, 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 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.

To control the size of each Partial_rows_log_event, a new system
variable is added: binlog_row_event_fragment_threshold. All
Partial_rows_log_events within the same group will have this size,
except for the last, which will take up the remaining length of the
underlying Rows_log_event.

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.

The re-assembly and execution of the original Rows_log_event on the
replica happens in Partial_rows_log_event::do_apply_event(). The rgi
is extended with a memory buffer that holds all data for the
original Rows_log_event. As each Partial_rows_log_event is
ingested, its Rows_log_event content is appended to this memory
buffer. Once the last fragment has added its content, a new
Rows_log_event is created using that buffer, and executed.

A new error message is added to indicate that the slave has received an
invalid stream of Partial_rows_log_events:
ER_PARTIAL_ROWS_LOG_EVENT_BAD_STREAM.

Note this commit only adds the server logic for fragmentic and
assembling events, the client logic (mysqlbinlog) is in the next
commit.

Alternative designs considered were:
  1. Alternative 1: Change the master-slave communication protocol
     such that the master would send events in chunks of size
     slave_max_allowed_packet. Though this is still a valid idea,
     and would solve the first problem described in this commit
     message, this would still leave the limitation that
     Rows_log_events could not exceed 4GB. Eventually, this change
     should still be addressed (e.g. in MDEV-37853), and for users
     of binlog-in-engine (MDEV-34705) which supports out-of-band
     binlogging of events, this MDEV-32570 work will be superseded.
  2. Alternative 2: Create a generic “Container_log_event” with the
     intention to embed various other types of event data for
     various purposes, with flags that describe the purpose of a
     given container. This seemed overboard, as there is already a
     generic Log_event framework that provides the necessary
     abstractions to fragment/reassemble events without adding in
     extra abstractions.
  3. Alternative 3: Add a flag to Rows_log_event with semantics to
     overwrite/correct the event_len field of the common event
     header to use a 64-bit field stored in the data_header of the
     Rows_log_event; and also do alternative 1, so the master would
     send the large (> 4GB) rows event in chunks. This approach
     would add too much complexity (changing both the binlogging
     and transport layer); as well as introduce inconsistency to
     the event definition (event_len and next_event_position would
     no longer have consistent meanings).

Reviewed-by: Hemant Dangi <hemant.dangi@mariadb.com>
Acked-by: Kristian Nielsen <knielsen@knielsen-hq.org>
Signed-off-by: Brandon Nesterenko <brandon.nesterenko@mariadb.com>
This commit is contained in:
Brandon Nesterenko
2025-05-15 13:24:06 -06:00
parent 8de56b4f25
commit 1eff7ddd81
12 changed files with 1192 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
sql/log.cc
View File

@@ -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;
}

113
sql/log_event.cc
View File

@@ -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<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),
@@ -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<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)

533
sql/log_event.h
View File

@@ -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<ulong>(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:
<table>
<caption>Post-Header for Partial_rows_log_event</caption>
<tr>
<th>Name</th>
<th>Format</th>
<th>Description</th>
</tr>
<tr>
<td>Sequence Number</td>
<td>4 byte unsigned integer</td>
<td>
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.
</td>
</tr>
<tr>
<td>Total Fragments</td>
<td>4 byte unsigned integer</td>
<td>
The total number of Partial_rows_log_events that exist to recreate the
original Rows_log_event
</td>
</tr>
<tr>
<td>Flags</td>
<td>1 byte for flags</td>
<td>
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).
</td>
</tr>
<tr>
<td>Reserved space (see flags bit 1)</td>
<td>0 bytes / 8 bytes</td>
<td>
Reserved bytes for storing optional data when indicated by flags
</td>
</tr>
</table>
The Body has the following components:
<table>
<caption>Body for Partial_rows_log_event</caption>
<tr>
<th>Name</th>
<th>Format</th>
<th>Description</th>
</tr>
<tr>
<td>content</td>
<td>String</td>
<td>
A fragment of the original Rows_log_event content.
</td>
</tr>
</table>
*/
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);

466
sql/log_event_server.cc
View File

@@ -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<uint32_t>(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<uint64_t>(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<uint64_t>(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<uint64_t>(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<uint32_t>((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;

5
sql/mysqld.cc
View File

@@ -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,

3
sql/mysqld.h
View File

@@ -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;

4
sql/rpl_rli.cc
View File

@@ -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(&current_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);

10
sql/rpl_rli.h
View File

@@ -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);

2
sql/share/errmsg-utf8.txt
View File

@@ -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"

23
sql/slave.cc
View File

@@ -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)) {

2
sql/slave.h
View File

@@ -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,

14
sql/sys_vars.cc
View File

@@ -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_var_uint,
PRIV_SET_SYSTEM_GLOBAL_VAR_SLAVE_NET_TIMEOUT>
Sys_slave_net_timeout(