Method sendQuery

src/Client/HedgedConnections.cpp:162–242

}

void HedgedConnections::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.");

    for (auto & offset_state : offset_states)
    {
        for (auto & replica : offset_state.replicas)
        {
            if (replica.connection->getServerRevision(timeouts) < DBMS_MIN_REVISION_WITH_CURRENT_AGGREGATION_VARIANT_SELECTION_METHOD)
            {
                disable_two_level_aggregation = true;
                break;
            }
        }
        if (disable_two_level_aggregation)
            break;
    }

    if (!disable_two_level_aggregation)
    {
        /// Tell hedged_connections_factory to skip replicas that doesn't support two-level aggregation.
        hedged_connections_factory.skipReplicasWithTwoLevelAggregationIncompatibility();
    }

    auto send_query = [this, timeouts, query, query_id, stage, client_info, with_pending_data, external_roles](ReplicaState & replica)
    {
        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 * offset_states.size());

        if (disable_two_level_aggregation)
        {
            /// 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;
        }

        const bool enable_offset_parallel_processing = context->canUseOffsetParallelReplicas();

        if (offset_states.size() > 1 && enable_offset_parallel_processing)
        {
            modified_settings[Setting::parallel_replicas_count] = offset_states.size();