ClickHouse源码分析：逻辑计划和Pipeline

基于clickhouse的版本：v21.3.4.35-lts

# 以TCP连接为例 select id from query_cache;
TCPHandler::runImpl
    BlockIO executeQuery // 解析query
        executeQueryImpl
            ast = parseQuery // 解析生成ast
                ...
            interpreter = InterpreterFactory::get // 根据ast实例化interpreter
            interpreter->execute() // InterpreterSelectWithUnionQuery::execute
                buildQueryPlan // 构建query paln 如 InterpreterSelectQuery::buildQueryPlan
                    InterpreterSelectQuery::executeImpl
                        executeFetchColumns
                            storage->read // StorageMergeTree::read --> MergeTreeDataSelectExecutor::read
                                StorageMergeTree::read
                                    MergeTreeDataSelectExecutor::read
                                        MergeTreeDataSelectExecutor::readFromParts // 根据一定条件选择需要读取的part，这里我们过滤掉如何选择的流程，直接去看怎么构建query plan
                                            spreadMarkRangesAmongStreams // 构建queryplan
                                                MergeTreeThreadSelectBlockInputProcessor
                                                unitePipes // 多个pipe绑定为一个pipe
                                                createPlanFromPipe // step = std::make_unique<ReadFromStorageStep>; plan->addStep(step); processor为MergeTreeThreadSelectBlockInputProcessor
                            query_plan.addStep // 增加step SettingQuotaAndLimits
                        executeExpression // 增加step ExpressionStep
                        executeProjection // 增加step ExpressionStep
                buildQueryPipeline // 构建pipeline
                    updatePipeline // ITransformingStep::updatePipeline-->ExpressionStep::transformPipeline-->addSimpleTransform-->collected_processors指向的processors和pipe.processors都增加processor; 即ExpressionTransform
            pipeline.addSimpleTransform // 增加processor LimitsCheckingTransform
    processOrdinaryQueryWithProcessors
        sendData(header); // 先向client发送header
        PullingAsyncPipelineExecutor executor(pipeline); // 构建一个执行器。 make_shared<LazyOutputFormat>
        PullingAsyncPipelineExecutor::pull
            threadFunction
                PipelineExecutor::execute
                    PipelineExecutor::executeImpl
                        initializeExecution
                            prepareProcessor
                                processor->prepare // 执行pushdata和pulldata操作
                        executeSingleThread
                            executeStepImpl // 根据线程号，弹出任务
                                addJob // 初始化node->job
                                node->job()
                                    executeJob
                                        processor->work() // 各个processor执行
                                            ...
    state.io.onFinish() // log和状态记录