Method commitPosition

src/Databases/MySQL/MaterializeMySQLSyncThread.cpp:627–653

}

void MaterializeMySQLSyncThread::commitPosition(const MySQLReplication::Position & position)
{
    if (position.binlog_name.empty())
        return;

    auto database = DatabaseCatalog::instance().getDatabase(database_name, getContext());
    auto materialized_mysql = dynamic_cast<DatabaseCloudMaterializedMySQL *>(database.get());
    if (!materialized_mysql)
        throw Exception("Expect CloudMaterializedMySQL database, but got " + database->getEngineName(), ErrorCodes::LOGICAL_ERROR);

    String binlog_meta_name = getNameForMaterializedBinlog(materialized_mysql->getDatabaseUUID(), assigned_materialized_table);

    auto binlog_metadata = getContext()->getCnchCatalog()->getMaterializedMySQLBinlogMetadata(binlog_meta_name);
    if (!binlog_metadata)
        throw Exception("Cannot get binlog meta from catalog for " + binlog_meta_name, ErrorCodes::LOGICAL_ERROR);

    if (binlog_metadata->binlog_file() != position.binlog_name
         || binlog_metadata->binlog_position() < position.binlog_pos)
    {
        binlog_metadata->set_binlog_file(position.binlog_name);
        binlog_metadata->set_binlog_position(position.binlog_pos);
        binlog_metadata->set_executed_gtid_set(position.gtid_sets.toString());

        LOG_DEBUG(log, "Try to commit position while no data in buffers now: {}", dumpPosition(position));
        getContext()->getCnchCatalog()->setMaterializedMySQLBinlogMetadata(binlog_meta_name, *binlog_metadata);
    }
}

void MaterializeMySQLSyncThread::tryCommitPosition(const MySQLReplication::Position & position)
{