Method initializePipeline

src/Processors/QueryPlan/ReadFromMergeTree.cpp:3573–4014

}

void ReadFromMergeTree::initializePipeline(QueryPipelineBuilder & pipeline, [[maybe_unused]] const BuildQueryPipelineSettings & settings)
{
    auto & result = getAnalysisResult();

    logPredicateStatistics(result);

    /// Filter ranges by 'bucket_id' parameter so that each distributed worker reads only its slice of the parts.
    if (distributed_read_bucket_count > 0 && settings.parameter_lookup)
    {
        /// Bucket over the coordinator-selected parts in a fixed order, so every worker partitions the
        /// same ordered list (replicas can have different local part layouts). A missing part is a
        /// retryable error rather than a silently divergent read.
        if (!distributed_read_part_names.empty())
        {
            std::unordered_map<String, RangesInDataPart> parts_by_name;
            for (auto & part : result.parts_with_ranges)
                parts_by_name.emplace(part.data_part->name, std::move(part));

            RangesInDataParts coordinator_parts;
            coordinator_parts.reserve(distributed_read_part_names.size());
            for (const auto & part_name : distributed_read_part_names)
            {
                auto it = parts_by_name.find(part_name);
                if (it == parts_by_name.end())
                    throw Exception(ErrorCodes::NO_SUCH_DATA_PART,
                        "Distributed read: part {} selected by the coordinator is not available on this replica "
                        "(diverged by merge or replication lag); retry the query", part_name);
                coordinator_parts.push_back(std::move(it->second));
            }
            result.parts_with_ranges = std::move(coordinator_parts);
        }

        const size_t bucket_id = parse<UInt64>(settings.parameter_lookup->getParameter("bucket_id").safeGet<String>());
        const size_t total_buckets = settings.parameter_lookup->getParameter("total_buckets").safeGet<UInt64>();

        size_t effective_bucket_index = bucket_id;
        RangesInDataParts filtered_parts;
        for (const auto & part : result.parts_with_ranges)
        {
            auto filtered_part = part;
            filtered_part.ranges = filterMarkRangesForBucket(part.ranges, effective_bucket_index, total_buckets);
            if (!filtered_part.ranges.empty())
                filtered_parts.push_back(std::move(filtered_part));
        }
        result.parts_with_ranges = std::move(filtered_parts);

        /// Cannot cache PREWHERE results when ranges are filtered by bucket_id.
        reader_settings.use_query_condition_cache = false;
    }

    if (enable_remove_parts_from_snapshot_optimization || query_info.isStream())
    {
        /// Do not keep data parts in snapshot.
        /// They are stored separately, and some could be released after PK analysis.
        /// Keep the underlying storage alive because part teardown still reaches
        /// `data_part->storage.getContext()`.
        auto stripped_snapshot_data = std::make_unique<MergeTreeData::SnapshotData>();
        if (const auto * snapshot_data = dynamic_cast<const MergeTreeData::SnapshotData *>(storage_snapshot->data.get()))