Repository: ClickHouse/ClickHouse
File: src/Interpreters/ConcurrentHashJoin.cpp (C++)

Code before:

    // If two-level maps are used, the probe stage uses only slot 0 (see joinBlock()),
    // and on build finish we merged buckets into slot 0 and copied the shared map/flags to all instances
    // but here we create multiple streams where each iterates over a subset of buckets (that originally belonged to that slot)
    // so hashtable in slot 0 looks like this:
    //┌─────────────────────────────────────────────────────┐
    //│ bucket[0] │ bucket[1] │ bucket[2] │ ... │ bucket[n] │
    //└─────────────────────────────────────────────────────┘
    if (hash_joins[0]->data->twoLevelMapIsUsed())
    {
        std::vector<IBlocksStreamPtr> streams;
        streams.reserve(slots);
        for (size_t i = 0; i < slots; ++i)
        {
            // Each slot iterates over buckets {i, i+slots, i+2*slots, ...}
            // This mirrors the bucket ownership during the parallel build phase
            // All slots share the same merged hash map (in slot 0), but each iterates different buckets
            std::lock_guard lock(hash_joins[0]->mutex);
            if (auto s = hash_joins[0]->data->getNonJoinedBlocks(left_sample_block, result_sample_block, max_block_size, i, slots))
                streams.push_back(std::move(s));
        }
        if (streams.empty())
            return {};
        if (streams.size() == 1)
            return streams[0];
        return std::make_shared<ConcatStreams>(std::move(streams));
    }
    else
    {
        // per-slot streams approach
        std::vector<IBlocksStreamPtr> streams;
        streams.reserve(slots);
        for (const auto & hash_join : hash_joins)
        {
            std::lock_guard lock(hash_join->mutex);
            if (hash_join->data->hasNonJoinedRows())
            {
                if (auto s = hash_join->data->getNonJoinedBlocks(left_sample_block, result_sample_block, max_block_size))
                    streams.push_back(std::move(s));
            }
        }
        if (streams.empty())
            return {};
        if (streams.size() == 1)
            return streams[0];
        return std::make_shared<ConcatStreams>(std::move(streams));
    }
}

template <typename HashTable>
static IColumn::Selector hashToSelector(const HashTable & hash_table, const BlockHashes & hashes, size_t num_shards)

Review comment (suggestion):
I didn't understand how it enables parallel output. We call `join->getNonJoinedBlocks` in `JoiningTransform::work`, and only a single `JoiningTransform` is responsible for handling this. Then in the same `JoiningTransform::work` we call `next` on the result. So won't we read from `ConcatStreams` in a single thread? Please correct me where I'm off.

Code after:
    /// no join keys (always false-condition), all right rows are non-joined, only stream 0 emits them
    if (table_join->getOnlyClause().key_names_right.empty())
    {
        if (stream_idx != 0)
            return {};
        std::lock_guard lock(hash_joins[0]->mutex);
        return hash_joins[0]->data->getNonJoinedBlocks(left_sample_block, result_sample_block, max_block_size);
    }

    /// Two-level maps: partition buckets across pipeline streams
    if (hash_joins[0]->data->twoLevelMapIsUsed())
    {
        std::lock_guard lock(hash_joins[0]->mutex);
        return hash_joins[0]->data->getNonJoinedBlocks(left_sample_block, result_sample_block, max_block_size, stream_idx, num_streams);
    }

    /// single-level maps - each slot has its own HashJoin, only stream 0 collects them
    if (stream_idx != 0)
        return {};

    std::vector<IBlocksStreamPtr> streams;
    streams.reserve(slots);
    for (const auto & hash_join : hash_joins)
    {
        std::lock_guard lock(hash_join->mutex);
        if (hash_join->data->hasNonJoinedRows())
        {
            if (auto s = hash_join->data->getNonJoinedBlocks(left_sample_block, result_sample_block, max_block_size))
                streams.push_back(std::move(s));
        }
    }
    if (streams.empty())
        return {};
    if (streams.size() == 1)
        return streams[0];
    return std::make_shared<ConcatStreams>(std::move(streams));
}

template <typename HashTable>
static IColumn::Selector hashToSelector(const HashTable & hash_table, const BlockHashes & hashes, size_t num_shards)
{
    assert(isPowerOf2(num_shards));
    const size_t num_rows = hashes.size();
    IColumn::Selector selector(num_rows);
    for (size_t i = 0; i < num_rows; ++i)
    {
        if constexpr (HasGetBucketFromHashMemberFunc<HashTable>)
            selector[i] = hash_table.getBucketFromHash(hashes[i]) & (num_shards - 1);
        else
            selector[i] = hashes[i] & (num_shards - 1);
    }