ClickHouse 源码解析（二）：SQL 的一生（中）

SELECT * FROM customer AS c
INNER JOIN order AS o ON c.C_CUSTKEY = o.C_CUSTKEY;

class Port
{
protected:
    /// Shared state of two connected ports.
    class State
    {
    public:
        struct Data
        {
            /// Note: std::variant can be used. But move constructor for it can't be inlined.
            Chunk chunk;
            std::exception_ptr exception;
        };
    private:
        std::atomic<Data *> data;
    }
}

/// Graph of executing pipeline.
class ExecutingGraph
{
public:
    /// Edge represents connection between OutputPort and InputPort.
    /// For every connection, two edges are created: direct and backward (it is specified by backward flag).
    struct Edge
    {
        /// Port numbers. They are same for direct and backward edges.
        uint64_t input_port_number;
        uint64_t output_port_number;
        
        /// Edge version is increased when port's state is changed (e.g. when data is pushed). See Port.h for details.
        /// To compare version with prev_version we can decide if neighbour processor need to be prepared.
        Port::UpdateInfo update_info;
    };

    /// Graph node. Represents single Processor.
    struct Node
    {
        /// Direct edges are for output ports, back edges are for input ports.
        Edges direct_edges;
        Edges back_edges;
    };

    Nodes nodes;
};

enum class Status
{
    /// Processor needs some data at its inputs to proceed.
    /// You need to run another processor to generate required input and then call 'prepare' again.
    NeedData,

    /// Processor cannot proceed because output port is full or not isNeeded().
    /// You need to transfer data from output port to the input port of another processor and then call 'prepare' again.
    PortFull,

    /// All work is done (all data is processed or all output are closed), nothing more to do.
    Finished,

    /// No one needs data on output ports.
    /// Unneeded,

    /// You may call 'work' method and processor will do some work synchronously.
    Ready,

    /// You may call 'schedule' method and processor will return descriptor.
    /// You need to poll this descriptor and call work() afterwards.
    Async,

    /// Processor wants to add other processors to pipeline.
    /// New processors must be obtained by expandPipeline() call.
    ExpandPipeline,
};

void ExecutingGraph::initializeExecution(Queue & queue)
{
    std::stack<uint64_t> stack;

    /// Add childless processors to stack.
    uint64_t num_processors = nodes.size();
    for (uint64_t proc = 0; proc < num_processors; ++proc)
    {
        /// 1.将不存在 direct_edges 的 Node 放入栈中,
        /// 在上面的例子中，就是将 LazyOutputFormat 放入栈.
        if (nodes[proc]->direct_edges.empty())
        {
            stack.push(proc);
            /// do not lock mutex, as this function is executed in single thread
            nodes[proc]->status = ExecutingGraph::ExecStatus::Preparing;
        }
    }

    Queue async_queue;

    while (!stack.empty())
    {
        uint64_t proc = stack.top();
        stack.pop();
        
        /// 2.更新 Node 状态.
        updateNode(proc, queue, async_queue);

        if (!async_queue.empty())
            throw Exception(ErrorCodes::LOGICAL_ERROR, "Async is only possible after work() call. Processor {}",
                            async_queue.front()->processor->getName());
    }
}

bool ExecutingGraph::updateNode(uint64_t pid, Queue & queue, Queue & async_queue)
{
    std::stack<Edge *> updated_edges;
    std::stack<uint64_t> updated_processors;
    updated_processors.push(pid);

    UpgradableMutex::ReadGuard read_lock(nodes_mutex);
    
    while (!updated_processors.empty() || !updated_edges.empty())
    {
        // 处理 update_edges 中 edge 指向的 node.
        if (updated_processors.empty()){...}
        
        // 处理 update_processors 中 node,更新 node 状态.
        if (!updated_processors.empty()){...}
    }
    return true;
}

if (!updated_processors.empty())
{
    pid = updated_processors.top();
    updated_processors.pop();

    /// In this method we have ownership on node.
    auto & node = *nodes[pid];
    {
        try
        {
            auto & processor = *node.processor;
            IProcessor::Status last_status = node.last_processor_status;
            // 1.调用 Node 对应 IProcessor::prepare() 方法尝试 pull 数据.
            IProcessor::Status status = processor.prepare(node.updated_input_ports, node.updated_output_ports);
            // 记录上一个 Processor 的 status.
            node.last_processor_status = status;
        }
        switch (node.last_processor_status)
        {
            // 2.NeedData 将 Node.status 从 Preparing -> Idle.
            case IProcessor::Status::NeedData:
            case IProcessor::Status::PortFull:
            {
                node.status = ExecutingGraph::ExecStatus::Idle;
                break;
            }
            ...
        }
        
        // 3.将 Node 相邻的待更新 Edge 放入 update_edges 这个栈中.
        if (!need_expand_pipeline)
        {        
            for (auto it = node.post_updated_output_ports.rbegin(); it != node.post_updated_output_ports.rend(); ++it)
            {
                auto * edge = static_cast<ExecutingGraph::Edge *>(*it);
                updated_edges.push(edge);
                edge->update_info.trigger();
            }
        
            for (auto it = node.post_updated_input_ports.rbegin(); it != node.post_updated_input_ports.rend(); ++it)
            {
                auto * edge = static_cast<ExecutingGraph::Edge *>(*it);
                updated_edges.push(edge);
                edge->update_info.trigger();
            }
        
            node.post_updated_input_ports.clear();
            node.post_updated_output_ports.clear();
        }
    }
}

IOutputFormat::Status IOutputFormat::prepare()
{
    // has_input 标识已经拉取到数据.
    if (has_input)
        return Status::Ready;
    
    // 默认就是 Main kind, Totals/Extremes 暂时不用管.
    for (auto kind : {Main, Totals, Extremes})
    {
        auto & input = getPort(kind);

        if (kind != Main && !input.isConnected())
            continue;

        if (input.isFinished())
            continue;
        
        // 更新相邻的 Edge，分别放入 
        // post_updated_input_ports、post_updated_output_ports
        input.setNeeded();
        
        // 查看 state 是否有数据,如果没有则返回 NeedData.
        if (!input.hasData())
            return Status::NeedData;

        // 如果 state 已经有数据,那么调用 pullData() 方法
        // 将数据拉取到 current_chunk 中,并将 has_input 设为 true.
        current_chunk = input.pull(true);
        current_block_kind = kind;
        has_input = true;
        return Status::Ready;
    }

    finished = true;

    if (!finalized)
        return Status::Ready;

    return Status::Finished;
}

if (updated_processors.empty())
{
    auto * edge = updated_edges.top();
    updated_edges.pop();

    /// Here we have ownership on edge, but node can be concurrently accessed.

    auto & node = *nodes[edge->to];

    std::unique_lock lock(node.status_mutex);

    ExecutingGraph::ExecStatus status = node.status;

    if (status != ExecutingGraph::ExecStatus::Finished)
    {
        if (edge->backward)
            node.updated_output_ports.push_back(edge->output_port_number);
        else
            node.updated_input_ports.push_back(edge->input_port_number);

        if (status == ExecutingGraph::ExecStatus::Idle)
        {
            node.status = ExecutingGraph::ExecStatus::Preparing;
            updated_processors.push(edge->to);
            stack_top_lock = std::move(lock);
        }
        else
            nodes[edge->to]->processor->onUpdatePorts();
    }
}

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

    /// Check can output.
    if (output.isFinished())
        return Status::Finished;

    if (!output.canPush())
        return Status::PortFull;
    
    // ISource 没有 input，所以会直接返回 Ready.
    if (!has_input)
        return Status::Ready;

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

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

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

    /// Now, we pushed to output, and it must be full.
    return Status::PortFull;
}

switch (node.last_processor_status)
{
    case IProcessor::Status::Ready:
    {
        node.status = ExecutingGraph::ExecStatus::Executing;
        queue.push(&node);
        break;
    }
}