Function createExchangeLookup

src/QueryPipeline/DistributedPlanExecutor.cpp:522–590

`query_id` must be the node-independent distributed query id: it keys the in-memory and streaming exchanges, so producers and consumers on different nodes (and the cleanup paths) must agree on it. It must not embed any node-local object-storage subpath, which would differ between nodes.

/// exchanges, so producers and consumers on different nodes (and the cleanup paths) must agree on it.
/// It must not embed any node-local object-storage subpath, which would differ between nodes.
ExchangeLookupPtr createExchangeLookup(
    const String & query_id,
    const ExchangeDescriptions & exchanges_,
    const ExchangeStreamSources & exchange_stream_sources,
    TemporaryFileLookupPtr temporary_files_,
    ContextPtr context)
{
    bool run_locally = context->getSettingsRef()[Setting::distributed_plan_execute_locally];
    if (run_locally)
    {
        LOG_DEBUG(getLogger("createExchangeLookup"), "`distributed_plan_execute_locally` setting is enabled, using in-memory queues for all exchanges");
        return std::make_shared<ExchangeViaChunks>(query_id);
    }

    auto persisted_exchanges = std::make_shared<ExchangeViaTemporaryFiles>(temporary_files_);

    bool has_streaming_exchange = false;
    for (const auto & [exchange_id, exchange] : exchanges_)
        if (exchange.kind == ExchangeDescription::Kind::Streaming)
        {
            has_streaming_exchange = true;
            break;
        }

    /// Persisted exchanges only need the temporary-file lookup, so a plan where every exchange
    /// is Persisted runs without a streaming transport (and on any platform). The streaming
    /// port and lookup are required only when the plan actually contains a Streaming exchange.
    if (!has_streaming_exchange)
    {
        UNUSED(exchange_stream_sources);
        return std::make_shared<AllKindsExchangeLookup>(exchanges_, persisted_exchanges, /*streaming_exchange_lookup=*/nullptr);
    }

#ifdef OS_LINUX
    auto streaming_exchange_port = context->getConfigRef().getUInt("distributed_query.streaming_exchange_port", 0);
    if (streaming_exchange_port == 0)
        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
            "Streaming exchange requires `distributed_query.streaming_exchange_port` to be configured; "
            "set the port, force `distributed_plan_force_exchange_kind = 'Persisted'`, or enable "
            "`distributed_plan_execute_locally` for in-process testing");
    if (streaming_exchange_port > 65535)
        throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER,
            "`distributed_query.streaming_exchange_port` must be in range 1..65535, got {}", streaming_exchange_port);

    /// The listener starts only when a listen host is also configured, so streaming peers are
    /// unreachable without one. Reject here instead of connecting to a listener that never started.
    if (getMultipleValuesFromConfig(context->getConfigRef(), "distributed_query", "streaming_exchange_listen_host").empty())
        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
            "Streaming exchange requires `distributed_query.streaming_exchange_listen_host` to be configured; "
            "set it, force `distributed_plan_force_exchange_kind = 'Persisted'`, or enable "
            "`distributed_plan_execute_locally` for in-process testing");

    /// A task from an older initiator (version 1) ships no per-stream ports; fall back to this
    /// node's configured exchange port to preserve the previous single-port behavior.
    ExchangeStreamSources sources_with_ports = exchange_stream_sources;
    for (auto & [stream, address] : sources_with_ports.stream_hosts)
        if (address.port == 0)
            address.port = static_cast<UInt16>(streaming_exchange_port);