mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-25 17:12:03 +00:00
5c221d123d
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()`
99 lines
3.5 KiB
C++
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>(>id), "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;
|
|
}
|