ClickHouse 详细代码分析文档

// src/Processors/IProcessor.h
class IProcessor
{
public:
    /// 处理器状态枚举
    enum class Status
    {
        NeedData,        /// 需要更多输入数据
        PortFull,        /// 输出端口已满
        Finished,        /// 处理完成
        Ready,           /// 准备处理
        Async,           /// 异步处理中
        ExpandPipeline   /// 需要扩展管道
    };

protected:
    InputPorts inputs;   /// 输入端口列表
    OutputPorts outputs; /// 输出端口列表

public:
    /// 核心处理方法 - 准备阶段
    virtual Status prepare() = 0;
    
    /// 核心处理方法 - 工作阶段
    virtual void work();
    
    /// 异步调度方法
    virtual int schedule();
    
    /// 管道扩展方法
    virtual Processors expandPipeline();
    
    /// 取消处理
    virtual void cancel() noexcept;
    
    /// 获取处理器名称
    virtual String getName() const = 0;
};

// src/Processors/IProcessor.cpp
IProcessor::IProcessor()
{
    // 分配处理器索引，用于调试和监控
    processor_index = CurrentThread::isInitialized() 
        ? CurrentThread::get().getNextPipelineProcessorIndex() 
        : 0;
}

IProcessor::IProcessor(InputPorts inputs_, OutputPorts outputs_) 
    : inputs(std::move(inputs_)), outputs(std::move(outputs_))
{
    // 建立端口与处理器的双向关联
    for (auto & port : inputs)
        port.processor = this;
    for (auto & port : outputs)
        port.processor = this;
        
    processor_index = CurrentThread::isInitialized() 
        ? CurrentThread::get().getNextPipelineProcessorIndex() 
        : 0;
}

void IProcessor::cancel() noexcept
{
    // 原子操作确保取消操作的线程安全性
    bool already_cancelled = is_cancelled.exchange(true, std::memory_order_acq_rel);
    if (already_cancelled)
        return;

    onCancel(); // 子类可重写此方法进行清理
}

// src/Processors/Executors/PipelineExecutor.h
class PipelineExecutor
{
public:
    /// 构造函数：构建执行图
    explicit PipelineExecutor(std::shared_ptr<Processors> & processors, QueryStatusPtr elem);
    
    /// 多线程执行管道
    void execute(size_t num_threads, bool concurrency_control);
    
    /// 单步执行（用于调试和控制）
    bool executeStep(std::atomic_bool * yield_flag = nullptr);

private:
    /// 执行图表示
    ExecutingGraphPtr graph;
    
    /// 任务队列
    ExecutorTasks tasks;
    
    /// 执行状态
    std::atomic<ExecutionStatus> execution_status{ExecutionStatus::EXECUTING};
};

// src/Processors/Executors/PipelineExecutor.cpp
void PipelineExecutor::execute(size_t num_threads, bool concurrency_control)
{
    checkTimeLimit();
    num_threads = std::max<size_t>(num_threads, 1);

    OpenTelemetry::SpanHolder span("PipelineExecutor::execute()");
    span.addAttribute("clickhouse.thread_num", num_threads);

    try
    {
        executeImpl(num_threads, concurrency_control);

        /// 记录所有LOGICAL_ERROR异常
        for (auto & node : graph->nodes)
            if (node->exception && getExceptionErrorCode(node->exception) == ErrorCodes::LOGICAL_ERROR)
                tryLogException(node->exception, log);

        /// 重新抛出第一个异常
        for (auto & node : graph->nodes)
            if (node->exception)
                std::rethrow_exception(node->exception);

        /// 处理执行线程中的异常
        tasks.rethrowFirstThreadException();
    }
    catch (...)
    {
        span.addAttribute(DB::ExecutionStatus::fromCurrentException());
        
#ifndef NDEBUG
        LOG_TRACE(log, "Exception while executing query. Current state:\n{}", dumpPipeline());
#endif
        throw;
    }

    finalizeExecution();
}

// src/Processors/Executors/ExecutingGraph.h
class ExecutingGraph
{
public:
    struct Node
    {
        ProcessorPtr processor;
        std::vector<Node *> direct_edges;     /// 直接依赖的节点
        std::vector<Node *> back_edges;       /// 反向依赖的节点
        
        /// 状态管理
        std::atomic<bool> status{false};
        std::exception_ptr exception;
        
        /// 调度信息
        std::atomic<UInt64> num_executed_jobs{0};
        std::atomic<UInt64> finish_time{0};
    };
    
    std::vector<std::unique_ptr<Node>> nodes;
    
    /// 构建执行图
    void buildGraph(Processors & processors);
    
    /// 获取可执行的节点
    std::vector<Node *> getReadyNodes();
    
    /// 更新节点状态
    void updateNodeStatus(Node * node, IProcessor::Status status);
};

void ExecutingGraph::buildGraph(Processors & processors)
{
    /// 1. 为每个处理器创建节点
    nodes.reserve(processors.size());
    std::unordered_map<IProcessor *, Node *> processor_to_node;
    
    for (auto & processor : processors)
    {
        auto node = std::make_unique<Node>();
        node->processor = processor;
        processor_to_node[processor.get()] = node.get();
        nodes.emplace_back(std::move(node));
    }
    
    /// 2. 建立节点间的依赖关系
    for (auto & node : nodes)
    {
        for (auto & input_port : node->processor->getInputs())
        {
            if (input_port.isConnected())
            {
                auto * output_processor = &input_port.getOutputPort().getProcessor();
                auto * output_node = processor_to_node[output_processor];
                
                /// 建立双向连接
                node->back_edges.push_back(output_node);
                output_node->direct_edges.push_back(node.get());
            }
        }
    }
    
    /// 3. 验证图的正确性
    validateGraph();
}

// src/Processors/Sources/SourceFromSingleChunk.h
class SourceFromSingleChunk : public ISource
{
public:
    SourceFromSingleChunk(Block header, Chunk chunk_)
        : ISource(std::move(header)), chunk(std::move(chunk_)) {}

    String getName() const override { return "SourceFromSingleChunk"; }

protected:
    Chunk generate() override
    {
        if (chunk)
        {
            auto res = std::move(chunk);
            chunk.clear();
            return res;
        }
        return {};
    }

private:
    Chunk chunk;
};

// ISource基类实现
class ISource : public IProcessor
{
public:
    ISource(Block header) : IProcessor({}, {OutputPort(std::move(header))}) {}

    Status prepare() override
    {
        if (finished)
        {
            output.finish();
            return Status::Finished;
        }

        if (output.isFinished())
        {
            finished = true;
            return Status::Finished;
        }

        if (!output.canPush())
            return Status::PortFull;

        if (!has_input)
            return Status::Ready;

        output.push(std::move(current_chunk));
        has_input = false;

        if (got_exception)
        {
            finished = true;
            output.finish();
            std::rethrow_exception(got_exception);
        }

        return Status::PortFull;
    }

    void work() override
    {
        try
        {
            current_chunk = generate();
            if (!current_chunk)
            {
                finished = true;
                return;
            }
            has_input = true;
        }
        catch (...)
        {
            finished = true;
            got_exception = std::current_exception();
        }
    }

protected:
    virtual Chunk generate() = 0;

private:
    OutputPort & output = outputs.front();
    bool finished = false;
    bool has_input = false;
    Chunk current_chunk;
    std::exception_ptr got_exception;
};

// src/Processors/Transforms/ExpressionTransform.h
class ExpressionTransform : public ISimpleTransform
{
public:
    ExpressionTransform(const Block & header_, ExpressionActionsPtr expression_)
        : ISimpleTransform(header_, expression_->getResultColumns().cloneEmpty(), false)
        , expression(std::move(expression_))
    {
        /// 检查表达式是否会改变行数
        const auto & actions = expression->getActions();
        for (const auto & action : actions)
        {
            if (action.node->type == ActionsDAG::ActionType::ARRAY_JOIN)
            {
                throw Exception(ErrorCodes::LOGICAL_ERROR, 
                    "ARRAY JOIN is not supported in ExpressionTransform");
            }
        }
    }

    String getName() const override { return "ExpressionTransform"; }

protected:
    void transform(Chunk & chunk) override
    {
        size_t num_rows = chunk.getNumRows();
        auto block = getInputPort().getHeader().cloneWithColumns(chunk.detachColumns());
        
        /// 执行表达式计算
        expression->execute(block, num_rows);
        
        chunk.setColumns(block.getColumns(), num_rows);
    }

private:
    ExpressionActionsPtr expression;
};

// ISimpleTransform基类实现
class ISimpleTransform : public IProcessor
{
public:
    ISimpleTransform(Block input_header, Block output_header, bool skip_empty_chunks_ = true)
        : IProcessor({InputPort(std::move(input_header))}, {OutputPort(std::move(output_header))})
        , input(inputs.front())
        , output(outputs.front())
        , skip_empty_chunks(skip_empty_chunks_) {}

    Status prepare() override
    {
        /// 检查输出端口状态
        if (output.isFinished())
        {
            input.close();
            return Status::Finished;
        }

        if (!output.canPush())
        {
            input.setNotNeeded();
            return Status::PortFull;
        }

        /// 检查输入端口状态
        if (has_output)
        {
            output.push(std::move(current_chunk));
            has_output = false;
        }

        if (finished_output)
        {
            output.finish();
            return Status::Finished;
        }

        if (has_input)
            return Status::Ready;

        if (input.isFinished())
        {
            finished_output = true;
            return Status::Ready;
        }

        input.setNeeded();
        if (!input.hasData())
            return Status::NeedData;

        current_chunk = input.pull(true);
        has_input = true;
        return Status::Ready;
    }

    void work() override
    {
        if (has_input)
        {
            transform(current_chunk);
            
            if (skip_empty_chunks && current_chunk.getNumRows() == 0)
            {
                has_input = false;
                return;
            }
            
            has_input = false;
            has_output = true;
        }
        else if (finished_input)
        {
            finished_output = true;
        }
    }

protected:
    virtual void transform(Chunk & chunk) = 0;

private:
    InputPort & input;
    OutputPort & output;
    
    Chunk current_chunk;
    bool has_input = false;
    bool has_output = false;
    bool finished_input = false;
    bool finished_output = false;
    
    const bool skip_empty_chunks = true;
};

// src/Processors/Sinks/SinkToStorage.h
class SinkToStorage : public ExceptionKeepingTransform
{
public:
    explicit SinkToStorage(const Block & header) : ExceptionKeepingTransform(header, {}) {}

    String getName() const override { return "SinkToStorage"; }

protected:
    void onConsume(Chunk chunk) override
    {
        if (!chunk)
            return;

        cur_chunk = std::move(chunk);
        consume(cur_chunk);
    }

    GenerateResult onGenerate() override
    {
        /// Sink不产生输出数据
        return {Chunk{}, false};
    }

    void onFinish() override
    {
        finalize();
    }

    /// 子类需要实现的方法
    virtual void consume(Chunk chunk) = 0;
    virtual void finalize() {}

private:
    Chunk cur_chunk;
};

// 具体的存储Sink实现示例
class MergeTreeSink : public SinkToStorage
{
public:
    MergeTreeSink(
        StorageMergeTree & storage_,
        const StorageMetadataPtr & metadata_snapshot_,
        size_t max_parts_per_block_,
        ContextPtr context_)
        : SinkToStorage(metadata_snapshot_->getSampleBlock())
        , storage(storage_)
        , metadata_snapshot(metadata_snapshot_)
        , max_parts_per_block(max_parts_per_block_)
        , context(context_) {}

    String getName() const override { return "MergeTreeSink"; }

protected:
    void consume(Chunk chunk) override
    {
        auto block = getHeader().cloneWithColumns(chunk.detachColumns());
        
        /// 写入数据到MergeTree
        storage.write(block, context);
    }

    void finalize() override
    {
        /// 完成写入，触发后台合并
        storage.flushAndPrepareForShutdown();
    }

private:
    StorageMergeTree & storage;
    StorageMetadataPtr metadata_snapshot;
    size_t max_parts_per_block;
    ContextPtr context;
};

// src/Storages/MergeTree/IMergeTreeDataPart.h
class IMergeTreeDataPart : public std::enable_shared_from_this<IMergeTreeDataPart>
{
public:
    /// 数据部分状态枚举
    enum class State
    {
        Temporary,       /// 临时状态，正在写入
        PreCommitted,    /// 预提交状态  
        Committed,       /// 已提交状态
        Outdated,        /// 过时状态，等待删除
        Deleting,        /// 正在删除
        DeleteOnDestroy  /// 析构时删除
    };
    
    /// 数据部分类型
    enum class Type
    {
        WIDE,           /// 宽格式（每列一个文件）
        COMPACT,        /// 紧凑格式（所有列在一个文件中）
        IN_MEMORY,      /// 内存格式
        UNKNOWN         /// 未知格式
    };

    /// 列大小统计
    struct ColumnSize
    {
        size_t marks = 0;                /// 标记数量
        size_t data_compressed = 0;      /// 压缩后大小
        size_t data_uncompressed = 0;    /// 压缩前大小
        
        void addToTotalSize(ColumnSize & total_size) const
        {
            total_size.marks += marks;
            total_size.data_compressed += data_compressed;
            total_size.data_uncompressed += data_uncompressed;
        }
    };

protected:
    /// 基本信息
    String name;                        /// 数据部分名称
    MergeTreePartInfo info;            /// 分区信息
    const MergeTreeData & storage;     /// 存储引擎引用
    VolumePtr volume;                  /// 存储卷

    /// 元数据
    mutable ColumnsDescription columns;              /// 列描述
    mutable SerializationInfoByName serialization_infos; /// 序列化信息
    mutable VersionMetadata version;                 /// 版本元数据

    /// 统计信息
    size_t rows_count = 0;             /// 行数
    size_t bytes_on_disk = 0;          /// 磁盘占用字节数
    mutable ColumnSizeByName columns_sizes; /// 各列大小统计

    /// 索引信息
    mutable IndexGranularity index_granularity;     /// 索引粒度
    size_t index_granularity_bytes = 0;             /// 索引粒度字节数

    /// 分区信息
    String partition_id;               /// 分区ID
    MergeTreePartition partition;      /// 分区值

    /// 校验和
    mutable Checksums checksums;       /// 文件校验和

    /// 状态管理
    mutable std::atomic<State> state{State::Temporary};
    mutable std::mutex state_mutex;

public:
    /// 获取数据读取器
    virtual std::shared_ptr<IMergeTreeReader> getReader(
        const NamesAndTypesList & columns_to_read,
        const StorageMetadataPtr & metadata_snapshot,
        const MarkRanges & mark_ranges,
        UncompressedCache * uncompressed_cache,
        MarkCache * mark_cache,
        const MergeTreeReaderSettings & reader_settings,
        const ValueSizeMap & avg_value_size_hints = {},
        const ReadBufferFromFileBase::ProfileCallback & profile_callback = {}) const = 0;

    /// 获取数据写入器
    virtual std::shared_ptr<IMergeTreeDataPartWriter> getWriter(
        const NamesAndTypesList & columns_list,
        const StorageMetadataPtr & metadata_snapshot,
        const std::vector<MergeTreeIndexPtr> & indices_to_recalc,
        const CompressionCodecPtr & default_codec_,
        const MergeTreeWriterSettings & writer_settings,
        const MergeTreeIndexGranularity & computed_index_granularity) const = 0;

    /// 加载元数据
    void loadColumnsChecksumsIndexes(bool require_columns_checksums, bool check_consistency);
    void loadIndex();
    void loadPartitionAndMinMaxIndex();
    void loadChecksums(bool require);
    void loadRowsCount();

    /// 校验数据完整性
    void checkConsistency(bool require_part_metadata) const;
    
    /// 计算各列大小
    void calculateEachColumnSizes(ColumnSizeByName & each_columns_size, ColumnSize & total_size) const;
};

// src/Storages/MergeTree/IMergeTreeReader.h
class IMergeTreeReader : private boost::noncopyable
{
public:
    using DeserializeBinaryBulkStateMap = std::unordered_map<std::string, ISerialization::DeserializeBinaryBulkStatePtr>;
    using FileStreams = std::map<std::string, std::unique_ptr<MergeTreeReaderStream>>;

    IMergeTreeReader(
        MergeTreeDataPartInfoForReaderPtr data_part_info_for_read_,
        const NamesAndTypesList & columns_,
        const VirtualFields & virtual_fields_,
        const StorageSnapshotPtr & storage_snapshot_,
        UncompressedCache * uncompressed_cache_,
        MarkCache * mark_cache_,
        const MarkRanges & all_mark_ranges_,
        const MergeTreeReaderSettings & settings_,
        const ValueSizeMap & avg_value_size_hints_ = ValueSizeMap{});

    /// 读取指定范围的数据行
    virtual size_t readRows(
        size_t from_mark, 
        size_t current_task_last_mark,
        bool continue_reading, 
        size_t max_rows_to_read, 
        Columns & res_columns) = 0;

    /// 预读取，用于异步IO优化
    virtual bool canReadIncompleteGranules() const = 0;

protected:
    /// 初始化文件流
    void addStreams(
        const NameAndTypePair & name_and_type,
        const ReadBufferFromFileBase::ProfileCallback & profile_callback);

    /// 读取数据流
    void readData(
        const NameAndTypePair & name_and_type,
        ColumnPtr & column,
        size_t from_mark,
        bool continue_reading,
        size_t current_task_last_mark,
        size_t max_rows_to_read,
        ISerialization::SubstreamsCache & cache,
        bool was_prefetched = false);

    MergeTreeDataPartInfoForReaderPtr data_part_info_for_read;
    const NamesAndTypesList columns;
    const VirtualFields virtual_fields;
    UncompressedCache * uncompressed_cache;
    MarkCache * mark_cache;
    MarkRanges all_mark_ranges;
    MergeTreeReaderSettings settings;
    
    /// 文件流管理
    FileStreams file_streams;
    
    /// 反序列化状态
    DeserializeBinaryBulkStateMap deserialize_binary_bulk_state_map;
};

// 具体的宽格式读取器实现
class MergeTreeReaderWide : public IMergeTreeReader
{
public:
    MergeTreeReaderWide(/* 参数列表 */) : IMergeTreeReader(/* 参数传递 */) {}

    size_t readRows(
        size_t from_mark, 
        size_t current_task_last_mark,
        bool continue_reading, 
        size_t max_rows_to_read, 
        Columns & res_columns) override
    {
        size_t read_rows = 0;
        
        /// 遍历所有需要读取的列
        for (size_t pos = 0; pos < columns.size(); ++pos)
        {
            const auto & name_and_type = columns[pos];
            
            if (!res_columns[pos])
                res_columns[pos] = name_and_type.type->createColumn();

            /// 从指定标记开始读取数据
            ISerialization::SubstreamsCache cache;
            readData(name_and_type, res_columns[pos], from_mark, continue_reading, 
                    current_task_last_mark, max_rows_to_read, cache);
        }

        /// 所有列应该读取相同的行数
        if (!res_columns.empty())
            read_rows = res_columns[0]->size();

        return read_rows;
    }

    bool canReadIncompleteGranules() const override { return true; }
};

// src/Storages/MergeTree/MergeTreeDataWriter.h
class MergeTreeDataWriter
{
public:
    MergeTreeDataWriter(MergeTreeData & data_) : data(data_), log(getLogger("MergeTreeDataWriter")) {}

    /// 写入数据块，返回创建的数据部分
    MergeTreeMutableDataPartPtr writeTempPart(
        BlockWithPartition & block_with_partition,
        const StorageMetadataPtr & metadata_snapshot,
        ContextPtr context);

private:
    /// 创建新的数据部分
    MergeTreeMutableDataPartPtr createPart(
        const String & part_name,
        const MergeTreeDataPartType & part_type,
        const MergeTreePartInfo & part_info,
        const VolumePtr & volume,
        const String & relative_path = "");

    /// 写入数据到磁盘
    void writeDataPart(
        MergeTreeMutableDataPartPtr & new_data_part,
        const Block & block,
        const StorageMetadataPtr & metadata_snapshot,
        ContextPtr context);

    MergeTreeData & data;
    LoggerPtr log;
};

// src/Storages/MergeTree/MergeTreeDataWriter.cpp
MergeTreeMutableDataPartPtr MergeTreeDataWriter::writeTempPart(
    BlockWithPartition & block_with_partition,
    const StorageMetadataPtr & metadata_snapshot,
    ContextPtr context)
{
    Block & block = block_with_partition.block;
    
    /// 1. 生成数据部分信息
    auto part_info = MergeTreePartInfo::fromPartName(
        data.getPartName(block_with_partition.partition), 
        data.format_version);
    
    /// 2. 选择存储卷
    auto volume = data.getStoragePolicy()->getVolume(0);
    
    /// 3. 创建新的数据部分
    auto new_data_part = createPart(
        data.getPartName(block_with_partition.partition),
        MergeTreeDataPartType::WIDE,
        part_info,
        volume);

    /// 4. 设置分区信息
    new_data_part->partition = std::move(block_with_partition.partition);
    new_data_part->minmax_idx->update(block, data.getMinMaxColumnsNames(metadata_snapshot->getPartitionKey()));

    /// 5. 写入数据到磁盘
    writeDataPart(new_data_part, block, metadata_snapshot, context);

    /// 6. 完成数据部分创建
    new_data_part->rows_count = block.rows();
    new_data_part->modification_time = time(nullptr);
    new_data_part->loadColumnsChecksumsIndexes(false, true);
    new_data_part->setBytesOnDisk(new_data_part->checksums.getTotalSizeOnDisk());

    return new_data_part;
}

void MergeTreeDataWriter::writeDataPart(
    MergeTreeMutableDataPartPtr & new_data_part,
    const Block & block,
    const StorageMetadataPtr & metadata_snapshot,
    ContextPtr context)
{
    /// 1. 创建数据写入器
    auto writer = new_data_part->getWriter(
        block.getNamesAndTypesList(),
        metadata_snapshot,
        {},  // indices_to_recalc
        data.getCompressionCodecForPart(new_data_part->info.level, new_data_part->info.mutation, context),
        MergeTreeWriterSettings(context->getSettingsRef(), data.getSettings()),
        computed_index_granularity);

    /// 2. 写入数据块
    writer->write(block);

    /// 3. 完成写入
    writer->finishDataSerialization(sync_on_insert);
    writer->finishPrimaryIndexSerialization(sync_on_insert);
    writer->finishSkipIndicesSerialization(sync_on_insert);

    /// 4. 计算校验和
    new_data_part->checksums = writer->releaseChecksums();
}

// src/Storages/MergeTree/MergeTreeDataSelectExecutor.h
class MergeTreeDataSelectExecutor
{
public:
    explicit MergeTreeDataSelectExecutor(const MergeTreeData & data_);

    /// 主要的查询执行方法
    QueryPlanPtr read(
        const Names & column_names,
        const StorageSnapshotPtr & storage_snapshot,
        const SelectQueryInfo & query_info,
        ContextPtr context,
        UInt64 max_block_size,
        size_t num_streams,
        PartitionIdToMaxBlockPtr max_block_numbers_to_read = nullptr,
        bool enable_parallel_reading = false) const;

    /// 从指定数据部分读取
    QueryPlanStepPtr readFromParts(
        RangesInDataParts parts,
        MergeTreeData::MutationsSnapshotPtr mutations_snapshot,
        const Names & column_names,
        const StorageSnapshotPtr & storage_snapshot,
        const SelectQueryInfo & query_info,
        ContextPtr context,
        UInt64 max_block_size,
        size_t num_streams,
        PartitionIdToMaxBlockPtr max_block_numbers_to_read = nullptr,
        ReadFromMergeTree::AnalysisResultPtr merge_tree_select_result_ptr = nullptr,
        bool enable_parallel_reading = false,
        std::shared_ptr<ParallelReadingExtension> extension_ = nullptr) const;

private:
    /// 选择需要读取的数据部分
    RangesInDataParts selectPartsToRead(
        const StorageSnapshotPtr & storage_snapshot,
        const SelectQueryInfo & query_info,
        ContextPtr context,
        PartitionIdToMaxBlockPtr max_block_numbers_to_read) const;

    /// 应用主键条件过滤
    void filterPartsByPrimaryKey(
        RangesInDataParts & parts,
        const StorageSnapshotPtr & storage_snapshot,
        const SelectQueryInfo & query_info,
        ContextPtr context) const;

    /// 应用跳数索引过滤
    void filterPartsBySkipIndexes(
        RangesInDataParts & parts,
        const StorageSnapshotPtr & storage_snapshot,
        const SelectQueryInfo & query_info,
        ContextPtr context) const;

    const MergeTreeData & data;
    LoggerPtr log;
};

// 查询执行核心逻辑
QueryPlanPtr MergeTreeDataSelectExecutor::read(
    const Names & column_names,
    const StorageSnapshotPtr & storage_snapshot,
    const SelectQueryInfo & query_info,
    ContextPtr context,
    UInt64 max_block_size,
    size_t num_streams,
    PartitionIdToMaxBlockPtr max_block_numbers_to_read,
    bool enable_parallel_reading) const
{
    /// 1. 选择需要读取的数据部分
    auto parts_with_ranges = selectPartsToRead(
        storage_snapshot, query_info, context, max_block_numbers_to_read);

    /// 2. 应用主键过滤
    filterPartsByPrimaryKey(parts_with_ranges, storage_snapshot, query_info, context);

    /// 3. 应用跳数索引过滤
    filterPartsBySkipIndexes(parts_with_ranges, storage_snapshot, query_info, context);

    /// 4. 创建查询计划
    auto query_plan = std::make_unique<QueryPlan>();
    
    if (parts_with_ranges.empty())
    {
        /// 没有数据需要读取，返回空结果
        auto header = storage_snapshot->getSampleBlockForColumns(column_names);
        auto read_nothing = std::make_unique<ReadNothingStep>(header);
        query_plan->addStep(std::move(read_nothing));
        return query_plan;
    }

    /// 5. 创建读取步骤
    auto read_step = readFromParts(
        std::move(parts_with_ranges),
        data.getMutationsSnapshot(query_info, context),
        column_names,
        storage_snapshot,
        query_info,
        context,
        max_block_size,
        num_streams,
        max_block_numbers_to_read,
        nullptr,
        enable_parallel_reading);

    query_plan->addStep(std::move(read_step));
    return query_plan;
}

// 数据部分选择的核心算法
RangesInDataParts MergeTreeDataSelectExecutor::selectPartsToRead(
    const StorageSnapshotPtr & storage_snapshot,
    const SelectQueryInfo & query_info,
    ContextPtr context,
    PartitionIdToMaxBlockPtr max_block_numbers_to_read) const
{
    RangesInDataParts parts_with_ranges;
    
    /// 1. 获取所有活跃的数据部分
    auto data_parts = data.getDataPartsVector();
    
    /// 2. 分区裁剪
    if (metadata_snapshot->hasPartitionKey())
    {
        const auto & partition_key = metadata_snapshot->getPartitionKey();
        auto partition_pruner = std::make_shared<PartitionPruner>(
            metadata_snapshot, query_info, context, false);
        
        /// 过滤不匹配分区条件的数据部分
        data_parts.erase(
            std::remove_if(data_parts.begin(), data_parts.end(),
                [&](const auto & part) 
                {
                    return !partition_pruner->canBePruned(*part);
                }),
            data_parts.end());
    }
    
    /// 3. 应用最大块号限制
    if (max_block_numbers_to_read)
    {
        data_parts.erase(
            std::remove_if(data_parts.begin(), data_parts.end(),
                [&](const auto & part)
                {
                    auto it = max_block_numbers_to_read->find(part->info.partition_id);
                    return it != max_block_numbers_to_read->end() 
                        && part->info.max_block > it->second;
                }),
            data_parts.end());
    }
    
    /// 4. 为每个数据部分创建标记范围
    for (const auto & part : data_parts)
    {
        RangesInDataPart ranges_in_part;
        ranges_in_part.data_part = part;
        
        /// 初始化为读取整个数据部分
        ranges_in_part.part_index_in_query = parts_with_ranges.size();
        ranges_in_part.ranges = MarkRanges{MarkRange{0, part->getMarksCount()}};
        
        parts_with_ranges.push_back(std::move(ranges_in_part));
    }
    
    return parts_with_ranges;
}

// 主键过滤实现
void MergeTreeDataSelectExecutor::filterPartsByPrimaryKey(
    RangesInDataParts & parts,
    const StorageSnapshotPtr & storage_snapshot,
    const SelectQueryInfo & query_info,
    ContextPtr context) const
{
    if (!storage_snapshot->getMetadataForQuery()->hasSortingKey())
        return;
        
    const auto & primary_key = storage_snapshot->getMetadataForQuery()->getPrimaryKey();
    
    /// 构建主键条件
    KeyCondition key_condition(query_info.query, context, primary_key.column_names, primary_key.expression);
    
    if (key_condition.alwaysUnknownOrTrue())
        return; /// 主键条件无法过滤任何数据
    
    /// 对每个数据部分应用主键过滤
    for (auto & part_with_ranges : parts)
    {
        if (part_with_ranges.ranges.empty())
            continue;
            
        auto & part = part_with_ranges.data_part;
        
        /// 加载主键索引
        part->loadIndex();
        
        /// 应用主键条件过滤标记范围
        MarkRanges filtered_ranges;
        
        for (const auto & range : part_with_ranges.ranges)
        {
            MarkRanges new_ranges = key_condition.mayBeTrueInRange(
                range.begin, range.end, part->index, primary_key.sample_block);
                
            filtered_ranges.insert(filtered_ranges.end(), 
                                 new_ranges.begin(), new_ranges.end());
        }
        
        part_with_ranges.ranges = std::move(filtered_ranges);
    }
    
    /// 移除空范围的数据部分
    parts.erase(
        std::remove_if(parts.begin(), parts.end(),
            [](const RangesInDataPart & part) { return part.ranges.empty(); }),
        parts.end());
}

// src/Storages/MergeTree/MergeTask.h
class MergeTask
{
public:
    /// 合并任务状态
    enum class State
    {
        NEED_PREPARE,
        NEED_EXECUTE,
        NEED_FINISH,
        SUCCESS
    };

    struct GlobalContext
    {
        MergeTreeData * data;
        StorageMetadataPtr metadata_snapshot;
        FutureMergedMutatedPartPtr future_part;
        MergeTreeData::MutableDataPartPtr new_data_part;
        
        /// 合并执行器
        std::unique_ptr<QueryPipelineBuilder> merging_pipeline;
        std::unique_ptr<PullingPipelineExecutor> merging_executor;
        
        /// 输出流
        std::unique_ptr<MergedBlockOutputStream> to;
        
        /// 统计信息
        size_t rows_written = 0;
        UInt64 watch_prev_elapsed = 0;
        
        /// 合并列表元素（用于监控）
        MergeListElement * merge_list_element_ptr = nullptr;
    };

    MergeTask(
        StorageMetadataPtr metadata_snapshot_,
        FutureMergedMutatedPartPtr future_part_,
        MergeTreeData * data_,
        MergeListElement * merge_list_element_ptr_,
        time_t time_of_merge_,
        ContextPtr context_,
        ReservationSharedPtr space_reservation_,
        bool deduplicate_,
        Names deduplicate_by_columns_,
        MergeTreeData::MergingParams merging_params_,
        MergeTreeTransactionPtr txn_,
        const String & suffix_ = "",
        bool need_prefix_ = true);

    /// 执行合并任务的一个步骤
    bool executeStep();

private:
    /// 准备合并
    bool prepare();
    
    /// 执行合并
    bool executeImpl();
    
    /// 完成合并
    bool finalize();

    std::shared_ptr<GlobalContext> global_ctx;
    State state{State::NEED_PREPARE};
};

// src/Storages/MergeTree/MergeTask.cpp
bool MergeTask::executeStep()
{
    switch (state)
    {
        case State::NEED_PREPARE:
            if (prepare())
            {
                state = State::NEED_EXECUTE;
                return true;
            }
            return false;

        case State::NEED_EXECUTE:
            if (executeImpl())
            {
                state = State::NEED_FINISH;
                return true;
            }
            return false;

        case State::NEED_FINISH:
            if (finalize())
            {
                state = State::SUCCESS;
                return false; /// 任务完成
            }
            return false;

        case State::SUCCESS:
            return false;
    }
    
    return false;
}

bool MergeTask::prepare()
{
    /// 1. 创建新的数据部分
    global_ctx->new_data_part = global_ctx->data->createPart(
        global_ctx->future_part->name,
        global_ctx->future_part->type,
        global_ctx->future_part->part_info,
        global_ctx->future_part->volume,
        global_ctx->future_part->relative_path);

    /// 2. 构建合并管道
    auto merging_pipeline = std::make_unique<QueryPipelineBuilder>();
    
    /// 为每个输入数据部分创建源
    for (const auto & part : global_ctx->future_part->parts)
    {
        auto source = std::make_shared<MergeTreeSequentialSource>(
            *global_ctx->data,
            global_ctx->metadata_snapshot,
            part,
            global_ctx->metadata_snapshot->getColumns().getNamesOfPhysical(),
            false, /// 不需要虚拟列
            true   /// 需要行号
        );
        
        merging_pipeline->addSource(std::move(source));
    }

    /// 3. 添加合并变换
    auto merging_transform = std::make_shared<MergingSortedTransform>(
        merging_pipeline->getHeader(),
        global_ctx->future_part->parts.size(),
        global_ctx->metadata_snapshot->getSortDescription(),
        global_ctx->data->merging_params.max_bytes_to_merge_at_max_space_in_pool);
    
    merging_pipeline->addTransform(std::move(merging_transform));

    /// 4. 创建输出流
    global_ctx->to = std::make_unique<MergedBlockOutputStream>(
        global_ctx->new_data_part,
        global_ctx->metadata_snapshot,
        global_ctx->metadata_snapshot->getColumns().getNamesOfPhysical(),
        CompressionCodecFactory::instance().get("NONE", {}),
        /// 其他参数...
    );

    /// 5. 创建执行器
    global_ctx->merging_executor = std::make_unique<PullingPipelineExecutor>(*merging_pipeline);
    global_ctx->merging_pipeline = std::move(merging_pipeline);

    return true;
}

bool MergeTask::executeImpl()
{
    /// 执行合并的时间限制
    UInt64 step_time_ms = global_ctx->data->getSettings()->background_task_preferred_step_execution_time_ms;
    
    Stopwatch watch;
    
    do
    {
        Block block;
        
        /// 从合并管道拉取数据块
        if (!global_ctx->merging_executor->pull(block))
        {
            /// 合并完成
            return true;
        }

        /// 写入合并后的数据块
        global_ctx->rows_written += block.rows();
        global_ctx->to->write(block);

        /// 更新最小最大索引
        if (global_ctx->data->merging_params.mode != MergeTreeData::MergingParams::Ordinary)
        {
            global_ctx->new_data_part->minmax_idx->update(
                block, 
                MergeTreeData::getMinMaxColumnsNames(global_ctx->metadata_snapshot->getPartitionKey()));
        }

        /// 更新统计信息
        if (global_ctx->merge_list_element_ptr)
        {
            global_ctx->merge_list_element_ptr->rows_written = global_ctx->rows_written;
            global_ctx->merge_list_element_ptr->bytes_written_uncompressed = 
                global_ctx->to->getBytesWritten();
        }

    } while (watch.elapsedMilliseconds() < step_time_ms);
    
    /// 时间片用完，但合并未完成
    return false;
}

bool MergeTask::finalize()
{
    /// 1. 完成数据写入
    global_ctx->to->finalizePart(global_ctx->new_data_part, false);
    
    /// 2. 设置数据部分属性
    global_ctx->new_data_part->rows_count = global_ctx->rows_written;
    global_ctx->new_data_part->modification_time = time(nullptr);
    
    /// 3. 加载元数据
    global_ctx->new_data_part->loadColumnsChecksumsIndexes(false, true);
    global_ctx->new_data_part->setBytesOnDisk(
        global_ctx->new_data_part->checksums.getTotalSizeOnDisk());
    
    /// 4. 提交新数据部分
    global_ctx->data->replaceParts(
        global_ctx->future_part->parts,
        {global_ctx->new_data_part},
        false);
    
    return true;
}

// src/Processors/Port.h
class Port
{
public:
    enum class State
    {
        NotNeeded,
        NeedData,
        HasData,
        Finished
    };

protected:
    State state = State::NotNeeded;
    IProcessor * processor = nullptr;
    
    /// 数据存储
    Chunk data;
    
    /// 连接信息
    Port * connected_port = nullptr;
    
public:
    /// 状态查询
    bool isConnected() const { return connected_port != nullptr; }
    bool hasData() const { return state == State::HasData; }
    bool isFinished() const { return state == State::Finished; }
    bool isNeeded() const { return state == State::NeedData; }
    
    /// 状态设置
    void setNeeded() { state = State::NeedData; }
    void setNotNeeded() { state = State::NotNeeded; }
    void finish() { state = State::Finished; }
    
    /// 数据传输
    void pushData(Chunk chunk)
    {
        chassert(state == State::NeedData);
        data = std::move(chunk);
        state = State::HasData;
    }
    
    Chunk pullData()
    {
        chassert(state == State::HasData);
        state = State::NotNeeded;
        return std::move(data);
    }
    
    /// 连接管理
    void connect(Port & other)
    {
        chassert(!isConnected() && !other.isConnected());
        connected_port = &other;
        other.connected_port = this;
    }
    
    void disconnect()
    {
        if (connected_port)
        {
            connected_port->connected_port = nullptr;
            connected_port = nullptr;
        }
    }
};

class InputPort : public Port
{
public:
    InputPort(Block header_) : header(std::move(header_)) {}
    
    /// 从连接的输出端口拉取数据
    Chunk pull(bool set_not_needed = false)
    {
        chassert(isConnected());
        chassert(hasData());
        
        auto chunk = connected_port->pullData();
        if (set_not_needed)
            setNotNeeded();
        return chunk;
    }
    
    /// 获取输出端口引用
    OutputPort & getOutputPort()
    {
        chassert(isConnected());
        return static_cast<OutputPort &>(*connected_port);
    }
    
    const Block & getHeader() const { return header; }

private:
    Block header; /// 数据块结构描述
};

class OutputPort : public Port
{
public:
    OutputPort(Block header_) : header(std::move(header_)) {}
    
    /// 向连接的输入端口推送数据
    void push(Chunk chunk)
    {
        chassert(isConnected());
        chassert(canPush());
        
        connected_port->pushData(std::move(chunk));
    }
    
    /// 检查是否可以推送数据
    bool canPush() const
    {
        chassert(isConnected());
        return connected_port->isNeeded();
    }
    
    /// 获取输入端口引用
    InputPort & getInputPort()
    {
        chassert(isConnected());
        return static_cast<InputPort &>(*connected_port);
    }
    
    const Block & getHeader() const { return header; }

private:
    Block header; /// 数据块结构描述
};

// src/QueryPipeline/QueryPipeline.h
class QueryPipeline
{
public:
    QueryPipeline() = default;
    QueryPipeline(QueryPipeline &&) = default;
    QueryPipeline & operator=(QueryPipeline &&) = default;
    
    /// 从单个处理器创建管道
    explicit QueryPipeline(std::shared_ptr<IProcessor> source);
    
    /// 从Pipe创建管道
    explicit QueryPipeline(Pipe pipe);
    
    /// 管道操作
    void addTransform(ProcessorPtr transform);
    void addSimpleTransform(const ProcessorGetter & getter);
    void addChains(std::vector<Chain> chains);
    
    /// 管道合并
    static QueryPipeline unitePipelines(
        std::vector<std::unique_ptr<QueryPipeline>> pipelines,
        size_t max_threads_limit = 0,
        Processors * collected_processors = nullptr);
    
    /// 执行管道
    PipelineExecutorPtr execute();
    
    /// 获取管道信息
    bool empty() const { return processors.empty(); }
    bool initialized() const { return !processors.empty() || !pipe.empty(); }
    size_t getNumStreams() const { return pipe.numOutputPorts(); }
    
    const Block & getHeader() const { return pipe.getHeader(); }
    
private:
    /// 处理器集合
    Processors processors;
    
    /// 管道表示
    Pipe pipe;
    
    /// 最大线程数
    size_t max_threads = 0;
};

// 管道连接的核心逻辑
void connectPorts(OutputPort & output, InputPort & input)
{
    /// 1. 检查端口兼容性
    if (!blocksHaveEqualStructure(output.getHeader(), input.getHeader()))
    {
        throw Exception(ErrorCodes::LOGICAL_ERROR,
            "Cannot connect ports with different block structures. "
            "Output header: {}, input header: {}",
            output.getHeader().dumpStructure(),
            input.getHeader().dumpStructure());
    }
    
    /// 2. 检查端口状态
    if (output.isConnected())
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Output port is already connected");
        
    if (input.isConnected())
        throw Exception(ErrorCodes::LOGICAL_ERROR, "Input port is already connected");
    
    /// 3. 建立连接
    output.connect(input);
}

// 自动连接处理器
void connectProcessors(IProcessor & left, IProcessor & right)
{
    auto & left_outputs = left.getOutputs();
    auto & right_inputs = right.getInputs();
    
    if (left_outputs.size() != right_inputs.size())
    {
        throw Exception(ErrorCodes::LOGICAL_ERROR,
            "Cannot connect processors: different number of ports. "
            "Left processor outputs: {}, right processor inputs: {}",
            left_outputs.size(), right_inputs.size());
    }
    
    auto left_it = left_outputs.begin();
    auto right_it = right_inputs.begin();
    
    for (; left_it != left_outputs.end(); ++left_it, ++right_it)
    {
        connectPorts(*left_it, *right_it);
    }
}