thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-30 05:30:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

ISSUES-4006 add MySQL replication GTIDEvent #4006

Browse Source
This commit is contained in:
BohuTANG 2020-08-17 09:23:54 +08:00
parent b4d18b1a52
commit e90679707a
2 changed files with 85 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
src/Core/MySQL/MySQLReplication.cpp
View File

@ -706,6 +706,28 @@ namespace MySQLReplication
}
}
void GTIDEvent::parseImpl(ReadBuffer & payload)
{
/// We only care uuid:seq_no parts assigned to GTID_NEXT.
payload.readStrict(reinterpret_cast<char *>(&commit_flag), 1);
payload.readStrict(reinterpret_cast<char *>(gtid.uuid), 16);
payload.readStrict(reinterpret_cast<char *>(&gtid.seq_no), 8);
/// Skip others.
payload.ignore(payload.available() - CHECKSUM_CRC32_SIGNATURE_LENGTH);
}
void GTIDEvent::dump(std::ostream & out) const
{
String dst36;
dst36.resize(36);
formatUUID(gtid.uuid, reinterpret_cast<UInt8 *>(dst36.data()));
auto gtid_next = dst36 + ":" + std::to_string(gtid.seq_no);
header.dump(out);
out << "GTID Next: " << gtid_next << std::endl;
}
void DryRunEvent::parseImpl(ReadBuffer & payload) { payload.ignore(header.event_size - EVENT_HEADER_LENGTH); }
void DryRunEvent::dump(std::ostream & out) const
@ -714,6 +736,43 @@ namespace MySQLReplication
out << "[DryRun Event]" << std::endl;
}
/// Update binlog name/position/gtid based on the event type.
void Position::update(BinlogEventPtr event)
{
switch (event->header.type)
{
case FORMAT_DESCRIPTION_EVENT: {
binlog_pos = event->header.log_pos;
break;
}
case ROTATE_EVENT: {
auto rotate = std::static_pointer_cast<RotateEvent>(event);
binlog_name = rotate->next_binlog;
binlog_pos = event->header.log_pos;
break;
}
case QUERY_EVENT: {
binlog_pos = event->header.log_pos;
break;
}
case XID_EVENT: {
binlog_pos = event->header.log_pos;
break;
}
case GTID_EVENT: {
auto gtid_event = std::static_pointer_cast<GTIDEvent>(event);
binlog_pos = event->header.log_pos;
gtid_sets.update(gtid_event->gtid);
break;
}
default: {
/// DryRun event.
binlog_pos = event->header.log_pos;
break;
}
}
}
void MySQLFlavor::readPayloadImpl(ReadBuffer & payload)
{
UInt16 header = static_cast<unsigned char>(*payload.position());
@ -736,17 +795,14 @@ namespace MySQLReplication
event = std::make_shared<FormatDescriptionEvent>();
event->parseHeader(payload);
event->parseEvent(payload);
position.updateLogPos(event->header.log_pos);
position.update(event);
break;
}
case ROTATE_EVENT: {
event = std::make_shared<RotateEvent>();
event->parseHeader(payload);
event->parseEvent(payload);
auto rotate = std::static_pointer_cast<RotateEvent>(event);
position.updateLogPos(event->header.log_pos);
position.updateLogName(rotate->next_binlog);
position.update(event);
break;
}
case QUERY_EVENT: {
@ -763,16 +819,15 @@ namespace MySQLReplication
break;
}
default:
position.updateLogPos(event->header.log_pos);
position.update(event);
}
break;
}
case XID_EVENT: {
event = std::make_shared<XIDEvent>();
event->parseHeader(payload);
event->parseEvent(payload);
position.updateLogPos(event->header.log_pos);
position.update(event);
break;
}
case TABLE_MAP_EVENT: {
@ -815,11 +870,18 @@ namespace MySQLReplication
event->parseEvent(payload);
break;
}
case GTID_EVENT: {
event = std::make_shared<GTIDEvent>();
event->parseHeader(payload);
event->parseEvent(payload);
position.update(event);
break;
}
default: {
event = std::make_shared<DryRunEvent>();
event->parseHeader(payload);
event->parseEvent(payload);
position.updateLogPos(event->header.log_pos);
position.update(event);
break;
}
}

16
src/Core/MySQL/MySQLReplication.h
View File

@ -1,6 +1,7 @@
#pragma once
#include <Core/Field.h>
#include <Core/MySQL/PacketsReplication.h>
#include <Core/MySQL/MySQLGtid.h>
#include <Core/Types.h>
#include <IO/ReadBuffer.h>
#include <IO/WriteBuffer.h>
@ -446,6 +447,17 @@ namespace MySQLReplication
MySQLEventType type() const override { return MYSQL_UPDATE_ROWS_EVENT; }
};
class GTIDEvent : public EventBase
{
public:
UInt8 commit_flag;
GTID gtid;
void dump(std::ostream & out) const override;
protected:
void parseImpl(ReadBuffer & payload) override;
};
class DryRunEvent : public EventBase
{
void dump(std::ostream & out) const override;
@ -459,11 +471,11 @@ namespace MySQLReplication
public:
UInt64 binlog_pos;
String binlog_name;
GTIDSets gtid_sets;
Position() : binlog_pos(0), binlog_name("") { }
Position(UInt64 binlog_pos_, const String & binlog_name_) : binlog_pos(binlog_pos_), binlog_name(binlog_name_) { }
void updateLogPos(UInt64 pos) { binlog_pos = pos; }
void updateLogName(String binlog) { binlog_name = std::move(binlog); }
void update(BinlogEventPtr event);
};
class IFlavor : public MySQLProtocol::IMySQLReadPacket