Method initializePipeline

src/QueryPlan/RemoteExchangeSourceStep.cpp:129–391

}

void RemoteExchangeSourceStep::initializePipeline(QueryPipeline & pipeline, const BuildQueryPipelineSettings & settings)
{
    auto current_tx_id = context->getCurrentTransactionID().toUInt64();
    if (!plan_segment)
        throw Exception("Should setPlanSegment before initializePipeline!", ErrorCodes::LOGICAL_ERROR);

    Pipe pipe;

    size_t source_num = 0;
    bool keep_order = context->getSettingsRef().exchange_enable_force_keep_order || context->getSettingsRef().enable_shuffle_with_order;
    if (!keep_order)
    {
        for (const auto & input : inputs)
        {
            if (input->needKeepOrder())
            {
                keep_order = input->needKeepOrder();
                break;
            }
        }
    }

    const Block & exchange_header = getOutputStream().header;
    Block source_header;
    if (keep_order)
        source_header = exchange_header;

    ExchangeTotalsSourcePtr totals_source;
    if (is_add_totals)
        totals_source = std::make_shared<ExchangeTotalsSource>(source_header);
    ExchangeExtremesSourcePtr extremes_source;
    if (is_add_extremes)
        extremes_source = std::make_shared<ExchangeExtremesSource>(source_header);
    auto enable_metrics = context->getSettingsRef().log_query_exchange;
    auto query_exchange_log = enable_metrics ? context->getQueryExchangeLog(): nullptr;
    auto register_mode
        = context->getSettingsRef().bsp_mode ? BrpcExchangeReceiverRegistryService::DISK_READER : BrpcExchangeReceiverRegistryService::BRPC;
    auto disk_exchange_mgr = context->getSettingsRef().bsp_mode ? context->getDiskExchangeDataManager() : nullptr;
    size_t local_queue_size = context->getSettingsRef().exchange_local_receiver_queue_size;
    size_t remote_queue_size = context->getSettingsRef().exchange_remote_receiver_queue_size;
    size_t multi_path_queue_size = context->getSettingsRef().exchange_multi_path_receiver_queue_size;
    std::shared_ptr<MemoryController> memory_controller;
    auto weak_segment_process_list_entry = context->getPlanSegmentProcessListEntry().lock();
    if (weak_segment_process_list_entry)
        memory_controller = weak_segment_process_list_entry->getMemoryController();

    for (const auto & input : inputs)
    {
        size_t write_plan_segment_id = input->getPlanSegmentId();
        size_t exchange_parallel_size = input->getExchangeParallelSize();
        UInt32 exchange_id = input->getExchangeId();
        UInt32 parallel_id = context->getPlanSegmentInstanceId().parallel_id;
        auto exchange_mode = input->getExchangeMode();
        //TODO: hack logic for BROADCAST/LOCAL_NO_NEED_REPARTITION/LOCAL_MAY_NEED_REPARTITION, we should remove this logic
        if (exchange_mode == ExchangeMode::LOCAL_NO_NEED_REPARTITION || exchange_mode == ExchangeMode::LOCAL_MAY_NEED_REPARTITION)
            parallel_id = 0;
        else if (exchange_mode == ExchangeMode::BROADCAST)
            exchange_parallel_size = 1;