ClickHouse/utils/check-mysql-binlog/main.cpp
Val Doroshchuk 5c221d123d MaterializedMySQL: Introduce MySQL Binlog Client
One binlog connection for many databases.

Suggesting to disable this feature by default for now. It should be explicitly enabled by SETTINGS use_binlog_client=1.
But if you would permanently enable it in MaterializedMySQLSettings, it should keep old behavior and all tests should pass too.

1. Introduced `IBinlog` and its impl to read the binlog events from socket - `BinlogFromSocket`, or file - `BinlogFromFile`. Based on prev impl of `EventBase` and the same old binlog parsers. It fully keeps BC with old version. Fixed `./check-mysql-binlog` to test new impl.
2. Introduced `BinlogEventsDispatcher`, it reads the event from the source `IBinlog` and sends it to currently attached `IBinlog` instances.
3. Introduced `BinlogClient`, which is used to group a list of `BinlogEventsDispatcher` by MySQL binlog connection which is defined by `user:password@host:port`. All dispatchers with the same binlog position should be merged to one.
4. Introduced `BinlogClientFactory`, which is a singleton and it is used to track all binlogs created over the instance.
5. Introduced `use_binlog_client` setting to `MaterializedMySQL`, which forces to reuse a `BinlogClient` if it already exists in `BinlogClientCatalog` or create new one. By default, it is disabled.
6. Introduced `max_bytes_in_binlog_queue` setting to define the limit of bytes in binlog's queue of events. If bytes in the queue increases this limit, `BinlogEventsDispatcher` will stop reading new events from source `IBinlog` until the space for new events will be freed.
7. Introduced `max_milliseconds_to_wait_in_binlog_queue` setting to define max ms to wait when the max bytes exceeded.
7. Introduced `max_milliseconds_to_wait_in_binlog_queue` setting to define max ms to wait when the max bytes exceeded.
8. Introduced `max_bytes_in_binlog_dispatcher_buffer` setting to define max bytes in the binlog dispatcher's buffer before it is flushed to attached binlogs.
9. Introduced `max_flush_milliseconds_in_binlog_dispatcher` setting to define max milliseconds in the binlog dispatcher's buffer to wait before it is flushed to attached binlogs.
10. Introduced `system.mysql_binlogs` system table, which shows a list of active binlogs.
11. Introduced `UnparsedRowsEvent` and `MYSQL_UNPARSED_ROWS_EVENT`, which defines that an event is not parsed and should be explicitly parsed later.
12. Fixed bug when not possible to apply DDL since syntax error or unsupported SQL.

@larspars is the author of following:
`GTIDSets::contains()`
`ReplicationHelper`
`shouldReconnectOnException()`
2024-01-03 15:26:09 +01:00

99 lines
3.5 KiB
C++

#include <IO/WriteBufferFromOStream.h>
#include <Databases/MySQL/MySQLBinlog.h>
#include <boost/program_options.hpp>
#include <iostream>
bool quit = false;
void signal_handler(int)
{
quit = true;
}
static void processBinlogFromFile(const std::string & bin_path, bool disable_checksum)
{
DB::MySQLReplication::BinlogFromFile binlog;
binlog.open(bin_path);
binlog.setChecksum(disable_checksum ? DB::MySQLReplication::IBinlog::NONE : DB::MySQLReplication::IBinlog::CRC32);
DB::MySQLReplication::BinlogEventPtr event;
while (binlog.tryReadEvent(event, /*timeout*/ 0) && !quit)
{
DB::WriteBufferFromOStream cout(std::cout);
event->dump(cout);
binlog.getPosition().dump(cout);
cout.finalize();
}
}
static void processBinlogFromSocket(const std::string & host, int port, const std::string & user, const std::string & password, const std::string & executed_gtid_set, bool disable_checksum)
{
DB::MySQLReplication::BinlogFromSocket binlog;
binlog.setChecksum(disable_checksum ? DB::MySQLReplication::IBinlog::NONE : DB::MySQLReplication::IBinlog::CRC32);
binlog.connect(host, port, user, password);
binlog.start(/*unique number*/ 42, executed_gtid_set);
DB::MySQLReplication::BinlogEventPtr event;
while (!quit)
{
if (binlog.tryReadEvent(event, /*timeout*/ 100))
{
if (event->header.type != DB::MySQLReplication::HEARTBEAT_EVENT)
{
DB::WriteBufferFromOStream cout(std::cout);
event->dump(cout);
binlog.getPosition().dump(cout);
cout.finalize();
}
}
}
}
int main(int argc, char ** argv)
{
(void)signal(SIGINT, signal_handler);
boost::program_options::options_description desc("Allowed options");
std::string host = "127.0.0.1";
int port = 3306;
std::string user = "root";
std::string password;
std::string gtid;
desc.add_options()
("help", "Produce help message")
("disable_checksum", "Disable checksums in binlog files.")
("binlog", boost::program_options::value<std::string>(), "Binlog file")
("host", boost::program_options::value<std::string>(&host)->default_value(host), "Host to connect")
("port", boost::program_options::value<int>(&port)->default_value(port), "Port number to connect")
("user", boost::program_options::value<std::string>(&user)->default_value(user), "User")
("password", boost::program_options::value<std::string>(&password), "Password")
("gtid", boost::program_options::value<std::string>(&gtid), "Executed gtid set");
try
{
boost::program_options::variables_map options;
boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), options);
boost::program_options::notify(options);
if (options.count("help") || (!options.count("binlog") && !options.count("gtid")))
{
std::cout << "Usage: " << argv[0] << std::endl;
std::cout << desc << std::endl;
return EXIT_FAILURE;
}
if (options.count("binlog"))
processBinlogFromFile(options["binlog"].as<std::string>(), options.count("disable_checksum"));
else
processBinlogFromSocket(host, port, user, password, gtid, options.count("disable_checksum"));
}
catch (std::exception & ex)
{
std::cerr << ex.what() << std::endl;
return EXIT_FAILURE;
}
return EXIT_SUCCESS;
}