Method read

src/Processors/QueryPlan/ReadFromMergeTree.cpp:856–901

}

Pipe ReadFromMergeTree::read(
    RangesInDataParts parts_with_range,
    const MergeTreeIndexBuildContextPtr & index_build_context,
    Names required_columns,
    ReadType read_type,
    size_t max_streams,
    size_t min_marks_for_concurrent_read,
    bool use_uncompressed_cache)
{
    const auto & settings = context->getSettingsRef();
    size_t sum_marks = parts_with_range.getMarksCountAllParts();

    const size_t total_query_nodes = is_parallel_reading_from_replicas
        ? std::min<size_t>(
              context->getClusterForParallelReplicas()->getShardsInfo().at(0).getAllNodeCount(),
              context->getSettingsRef()[Setting::max_parallel_replicas])
        : 1;

    PoolSettings pool_settings{
        .threads = max_streams,
        .sum_marks = sum_marks,
        .min_marks_for_concurrent_read = min_marks_for_concurrent_read,
        .preferred_block_size_bytes = settings[Setting::preferred_block_size_bytes],
        .use_uncompressed_cache = use_uncompressed_cache,
        .use_const_size_tasks_for_remote_reading = settings[Setting::merge_tree_use_const_size_tasks_for_remote_reading],
        .total_query_nodes = total_query_nodes,
    };

    if (read_type == ReadType::ParallelReplicas)
        return readFromPoolParallelReplicas(
            std::move(parts_with_range), index_build_context, std::move(required_columns), std::move(pool_settings));

    /// Reading from default thread pool is beneficial for remote storage because of new prefetches.
    if (read_type == ReadType::Default && (max_streams > 1 || checkAllPartsOnRemoteFS(parts_with_range)))
        return readFromPool(
            std::move(parts_with_range), index_build_context, std::move(required_columns), std::move(pool_settings));

    auto pipe = readInOrder(parts_with_range, index_build_context, required_columns, pool_settings, read_type, /*limit=*/0);

    /// Use ConcatProcessor to concat sources together.
    /// It is needed to read in parts order (and so in PK order) if single thread is used.
    if (read_type == ReadType::Default && pipe.numOutputPorts() > 1)
        pipe.addTransform(std::make_shared<ConcatProcessor>(pipe.getSharedHeader(), pipe.numOutputPorts()));

    return pipe;
}

namespace
{