Method sendQuery

src/Client/MultiplexedConnections.cpp:142–218

}

void MultiplexedConnections::sendQuery(
    const ConnectionTimeouts & timeouts,
    const String & query,
    const String & query_id,
    UInt64 stage,
    ClientInfo & client_info,
    bool with_pending_data,
    const std::vector<String> & external_roles)
{
    std::lock_guard lock(cancel_mutex);

    if (sent_query)
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Query already sent.");

    Settings modified_settings = settings;

    /// Queries in foreign languages are transformed to ClickHouse-SQL. Ensure the setting before sending.
    modified_settings[Setting::dialect] = Dialect::clickhouse;
    modified_settings[Setting::dialect].changed = false;

    modified_settings[Setting::interactive_delay] = scaleInteractiveDelayByFanout(
        modified_settings[Setting::interactive_delay],
        distributed_fanout * replica_states.size());

    for (auto & replica : replica_states)
    {
        if (!replica.connection)
            throw Exception(ErrorCodes::LOGICAL_ERROR, "MultiplexedConnections: Internal error");

        if (replica.connection->getServerRevision(timeouts) < DBMS_MIN_REVISION_WITH_CURRENT_AGGREGATION_VARIANT_SELECTION_METHOD)
        {
            /// Disable two-level aggregation due to version incompatibility.
            modified_settings[Setting::group_by_two_level_threshold] = 0;
            modified_settings[Setting::group_by_two_level_threshold_bytes] = 0;
        }
    }

    if (replica_info)
    {
        client_info.collaborate_with_initiator = true;
        client_info.number_of_current_replica = replica_info->number_of_current_replica;
    }

    /// FIXME: Remove once we will make `allow_experimental_analyzer` obsolete setting.
    /// Make the analyzer being set, so it will be effectively applied on the remote server.
    /// In other words, the initiator always controls whether the analyzer enabled or not for
    /// all servers involved in the distributed query processing.
    modified_settings.set("allow_experimental_analyzer", static_cast<bool>(modified_settings[Setting::allow_experimental_analyzer]));

    const bool enable_offset_parallel_processing = context->canUseOffsetParallelReplicas();

    size_t num_replicas = replica_states.size();
    chassert(num_replicas > 0);
    if (num_replicas > 1)
    {
        if (enable_offset_parallel_processing)
            /// Use multiple replicas for parallel query processing.
            modified_settings[Setting::parallel_replicas_count] = num_replicas;