Method sendBatch

src/Storages/Distributed/DistributedAsyncInsertBatch.cpp:244–310

}

void DistributedAsyncInsertBatch::sendBatch(const SettingsChanges & settings_changes)
{
    IConnectionPool::Entry connection;
    std::unique_ptr<RemoteInserter> remote;
    bool compression_expected = false;

    /// Since the batch is sent as a whole (in case of failure, the whole batch
    /// will be repeated), we need to mark the whole batch as failed in case of
    /// error).
    std::vector<OpenTelemetry::TracingContextHolderPtr> tracing_contexts;
    UInt64 batch_start_time = clock_gettime_ns();

    try
    {
        for (const auto & file : files)
        {
            ReadBufferFromFile in(file);
            const auto & distributed_header = DistributedAsyncInsertHeader::read(in, parent.log);

            tracing_contexts.emplace_back(distributed_header.createTracingContextHolder(
                __PRETTY_FUNCTION__,
                parent.storage.getContext()->getOpenTelemetrySpanLog()));
            tracing_contexts.back()->root_span.addAttribute("clickhouse.distributed_batch_start_time", batch_start_time);

            if (!remote)
            {
                Settings insert_settings = *distributed_header.insert_settings;
                insert_settings.applyChanges(settings_changes);

                auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(insert_settings);
                auto results = parent.pool->getManyCheckedForInsert(timeouts, insert_settings, PoolMode::GET_ONE, parent.storage.remote_storage.getQualifiedName());
                auto result = parent.pool->getValidTryResult(results, insert_settings[Setting::distributed_insert_skip_read_only_replicas]);
                connection = std::move(result.entry);
                compression_expected = connection->getCompression() == Protocol::Compression::Enable;

                LOG_DEBUG(parent.log, "Sending a batch of {} files to {} ({} rows, {} bytes).",
                    files.size(),
                    connection->getDescription(),
                    formatReadableQuantity(total_rows),
                    formatReadableSizeWithBinarySuffix(total_bytes));

                remote = std::make_unique<RemoteInserter>(*connection, timeouts,
                    distributed_header.insert_query,
                    insert_settings,
                    distributed_header.client_info);
                remote->initialize();
            }
            writeRemoteConvert(distributed_header, *remote, compression_expected, in, parent.log);
        }

        if (remote)
            remote->onFinish();
    }
    catch (...)
    {
        try
        {
            for (auto & tracing_context : tracing_contexts)