Method commit

src/Databases/MySQL/MaterializeMySQLSyncThread.cpp:781–819

}

void MaterializeMySQLSyncThread::Buffers::commit(ContextPtr context, const MySQLReplication::Position & position)
{
    try
    {
        if (position.binlog_name.empty())
            throw Exception("binlog is empty when try to committing data for MaterializedMySQL", ErrorCodes::LOGICAL_ERROR);

        MySQLBinLogInfo binlog;
        binlog.binlog_file = position.binlog_name;
        binlog.binlog_position = position.binlog_pos;
        binlog.executed_gtid_set = position.gtid_sets.toString();

        /// XXX: Here we can just support for one table to flush
        for (auto & table_name_and_buffer : data)
        {
            auto query_context = MaterializedMySQL::createQueryContext(context);

            prepareCommit(query_context, table_name_and_buffer.first + "_" + table_suffix, binlog);

            OneBlockInputStream input(table_name_and_buffer.second->first);
            BlockOutputStreamPtr out = getTableOutput(database, table_name_and_buffer.first + "_" + table_suffix, query_context, true);
            Stopwatch watch;
            copyData(input, *out);
            LOG_DEBUG(&Poco::Logger::get("MaterializeMySQLThread ({})"), "Copied {} rows and elapsed {} ms",
                table_name_and_buffer.first, table_name_and_buffer.second->first.rows(), watch.elapsedMilliseconds());
        }

        data.clear();
        max_block_rows = 0;
        max_block_bytes = 0;
        total_blocks_rows = 0;
        total_blocks_bytes = 0;
    }
    catch (...)
    {
        data.clear();
        throw;
    }
}

MaterializeMySQLSyncThread::Buffers::BufferAndUniqueColumnsPtr MaterializeMySQLSyncThread::Buffers::getTableDataBuffer(
    const String & table_name, ContextPtr context)