Function executeQuery

src/Interpreters/ClusterProxy/executeQuery.cpp:340–525

void executeQuery(
    QueryPlan & query_plan,
    SharedHeader header,
    QueryProcessingStage::Enum processed_stage,
    const StorageID & main_table,
    const ASTPtr & table_func_ptr,
    SelectStreamFactory & stream_factory,
    LoggerPtr log,
    ContextPtr context,
    const SelectQueryInfo & query_info,
    const ExpressionActionsPtr & sharding_key_expr,
    const std::string & sharding_key_column_name,
    const DistributedSettings & distributed_settings,
    AdditionalShardFilterGenerator shard_filter_generator,
    bool is_remote_function)
{
    const Settings & settings = context->getSettingsRef();

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

    const ClusterPtr & not_optimized_cluster = query_info.cluster;

    std::vector<QueryPlanPtr> plans;
    SelectStreamFactory::Shards remote_shards;

    auto cluster = query_info.getCluster();
    auto new_context = updateSettingsAndClientInfoForCluster(*cluster, is_remote_function, context,
        settings, main_table, query_info.additional_filter_ast, log, &distributed_settings);
    if (context->getSettingsRef()[Setting::allow_experimental_parallel_reading_from_replicas].value
        && context->getSettingsRef()[Setting::allow_experimental_parallel_reading_from_replicas].value
           != new_context->getSettingsRef()[Setting::allow_experimental_parallel_reading_from_replicas].value)
    {
        LOG_TRACE(
            log,
            "Parallel reading from replicas is disabled for cluster. There are no shards with more than 1 replica: cluster={}",
            cluster->getName());
    }

    new_context->increaseDistributedDepth();

    const size_t shards = cluster->getShardCount();
    ProfileEvents::increment(ProfileEvents::Shards, shards);

    /// Tracker is shared between local-missing-table skip path in SelectStreamFactory and
    /// remote unavailable-shard skip path in ReadFromRemote so max_skip_unavailable_shards_num
    /// and max_skip_unavailable_shards_ratio are enforced uniformly across both paths.
    UnavailableShardTrackerPtr unavailable_shard_tracker;
    {
        const auto & new_settings_ref = new_context->getSettingsRef();
        if (new_settings_ref[Setting::skip_unavailable_shards])
        {
            size_t max_num = new_settings_ref[Setting::max_skip_unavailable_shards_num];
            Float64 max_ratio = static_cast<double>(new_settings_ref[Setting::max_skip_unavailable_shards_ratio]);
            if (max_num > 0 || max_ratio > 0)
                unavailable_shard_tracker = std::make_shared<UnavailableShardTracker>(shards, max_num, max_ratio);
        }
    }