Function executeQuery

src/Interpreters/ClusterProxy/executeQuery.cpp:172–298

}

void executeQuery(
    QueryPlan & query_plan,
    IStreamFactory & stream_factory, Poco::Logger * log,
    const ASTPtr & query_ast, ContextPtr context, const SelectQueryInfo & query_info,
    const ExpressionActionsPtr & sharding_key_expr,
    const std::string & sharding_key_column_name,
    const ClusterPtr & not_optimized_cluster)
{
    assert(log);

    const Settings & settings = context->getSettingsRef();

    if (settings.max_distributed_depth && context->getClientInfo().distributed_depth > settings.max_distributed_depth)
        throw Exception("Maximum distributed depth exceeded", ErrorCodes::TOO_LARGE_DISTRIBUTED_DEPTH);

    std::vector<QueryPlanPtr> plans;
    Pipes remote_pipes;
    Pipes delayed_pipes;

    auto new_context = updateSettingsForCluster(*query_info.getCluster(), context, settings, log);

    new_context->getClientInfo().distributed_depth += 1;

    ThrottlerPtr user_level_throttler;
    if (auto * process_list_element = context->getProcessListElement())
        user_level_throttler = process_list_element->getUserNetworkThrottler();

    /// Network bandwidth limit, if needed.
    ThrottlerPtr throttler;
    if (settings.max_network_bandwidth || settings.max_network_bytes)
    {
        throttler = std::make_shared<Throttler>(
                settings.max_network_bandwidth,
                settings.max_network_bytes,
                "Limit for bytes to send or receive over network exceeded.",
                user_level_throttler);
    }
    else
        throttler = user_level_throttler;

    ASTPtr rewrite_ast = query_ast;

    std::set<UInt32> skip_shards;
    std::vector<String> skip_shards_str;
    String skip_shard_list = settings.skip_shard_list;
    boost::split(skip_shards_str, skip_shard_list, boost::is_any_of(" ,"));
    for (auto& shard : skip_shards_str)
    {
        if (shard.empty()) continue; // default empty setting
        try
        {
            skip_shards.insert(std::stoi(shard));
        }
        catch(...)
        {
            throw Exception("skip_shard_list setting wrong " + skip_shard_list, ErrorCodes::UNKNOWN_SETTING);
        }
    }