Method ~ConcurrentHashJoin

src/Interpreters/ConcurrentHashJoin.cpp:236–277

}

ConcurrentHashJoin::~ConcurrentHashJoin()
{
    try
    {
        if (!build_phase_finished || !hash_joins[0]->data->twoLevelMapIsUsed())
            return;

        updateStatistics(hash_joins, stats_collecting_params);

        for (size_t i = 0; i < slots; ++i)
        {
            // Hash tables destruction may be very time-consuming.
            // Without the following code, they would be destroyed in the current thread (i.e. sequentially).
            pool->scheduleOrThrow(
                [join = hash_joins[0], i, this, thread_group = CurrentThread::getGroup()]()
                {
                    ThreadGroupSwitcher switcher(thread_group, ThreadName::CONCURRENT_JOIN);

                    auto clear_space_in_buckets = [&](auto & maps, HashJoin::Type type, size_t idx)
                    {
                        APPLY_TO_MAP(
                            INVOKE_WITH_MAP,
                            type,
                            maps,
                            [&](auto & map)
                            {
                                for (size_t j = idx; j < map.NUM_BUCKETS; j += slots)
                                    map.impls[j].clearAndShrink();
                            })
                    };
                    const auto & right_data = getData(join);
                    std::visit([&](auto & maps) { return clear_space_in_buckets(maps, right_data->type, i); }, right_data->maps.at(0));
                });
        }
        pool->wait();
    }
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);
        pool->wait();
    }
}

bool ConcurrentHashJoin::addBlockToJoin(const Block & right_block_, bool check_limits)
{