Merge pull request ClickHouse#76700 from ClickHouse/backport/25.1/75149

Backport ClickHouse#75149 to 25.1: Fix double preallocation in `ConcurrentHashJoin` in case of swap join sides

Author: robot-clickhouse-ci-2

src/Interpreters/ConcurrentHashJoin.cpp

@@ -101,18 +101,13 @@ HashJoin::RightTableDataPtr getData(const std::shared_ptr<ConcurrentHashJoin::In
     return join->data->getJoinedData();
 }
 
-void reserveSpaceInHashMaps(
-    const std::vector<std::shared_ptr<ConcurrentHashJoin::InternalHashJoin>> & hash_joins,
-    ThreadPool * pool,
-    const StatsCollectingParams & stats_collecting_params,
-    size_t slots)
+void reserveSpaceInHashMaps(HashJoin & hash_join, size_t ind, const StatsCollectingParams & stats_collecting_params, size_t slots)
 {
     if (auto hint = getSizeHint(stats_collecting_params, slots))
     {
         /// Hash map is shared between all `HashJoin` instances, so the `median_size` is actually the total size
         /// we need to preallocate in all buckets of all hash maps.
         const size_t reserve_size = hint->median_size;
-        ProfileEvents::increment(ProfileEvents::HashJoinPreallocatedElementsInHashTables, reserve_size);
 
         /// Each `HashJoin` instance will "own" a subset of buckets during the build phase. Because of that
         /// we preallocate space only in the specific buckets of each `HashJoin` instance.
@@ -129,25 +124,9 @@ void reserveSpaceInHashMaps(
                 })
         };
 
-        for (size_t i = 0; i < slots; ++i)
-        {
-            pool->scheduleOrThrow(
-                [&, i, thread_group = CurrentThread::getGroup()]()
-                {
-                    SCOPE_EXIT_SAFE({
-                        if (thread_group)
-                            CurrentThread::detachFromGroupIfNotDetached();
-                    });
-
-                    if (thread_group)
-                        CurrentThread::attachToGroupIfDetached(thread_group);
-                    setThreadName("ConcurrentJoin");
-
-                    const auto & right_data = getData(hash_joins[i]);
-                    std::visit([&](auto & maps) { return reserve_space_in_buckets(maps, right_data->type, i); }, right_data->maps.at(0));
-                });
-        }
-        pool->wait();
+        const auto & right_data = hash_join.getJoinedData();
+        std::visit([&](auto & maps) { return reserve_space_in_buckets(maps, right_data->type, ind); }, right_data->maps.at(0));
+        ProfileEvents::increment(ProfileEvents::HashJoinPreallocatedElementsInHashTables, reserve_size / slots);
     }
 }
 
@@ -220,12 +199,6 @@ ConcurrentHashJoin::ConcurrentHashJoin(
                 });
         }
         pool->wait();
-
-        if (!hash_joins[0]->data->twoLevelMapIsUsed())
-            /// Means fixed-size hash map is used
-            return;
-
-        reserveSpaceInHashMaps(hash_joins, pool.get(), stats_collecting_params, slots);
     }
     catch (...)
     {
@@ -316,6 +289,12 @@ bool ConcurrentHashJoin::addBlockToJoin(const Block & right_block_, bool check_l
                 if (!lock.owns_lock())
                     continue;
 
+                if (!hash_join->space_was_preallocated && hash_join->data->twoLevelMapIsUsed())
+                {
+                    reserveSpaceInHashMaps(*hash_join->data, i, stats_collecting_params, slots);
+                    hash_join->space_was_preallocated = true;
+                }
+
                 bool limit_exceeded = !hash_join->data->addBlockToJoin(dispatched_block, check_limits);
 
                 dispatched_block = {};
@@ -636,7 +615,7 @@ void ConcurrentHashJoin::onBuildPhaseFinish()
 {
     for (auto & hash_join : hash_joins)
     {
-        // It cannot be called concurrently with other IJoin methods
+        // `onBuildPhaseFinish` cannot be called concurrently with other IJoin methods, so we don't need a lock to access internal joins.
         hash_join->data->onBuildPhaseFinish();
     }
 

src/Interpreters/ConcurrentHashJoin.h

@@ -71,7 +71,6 @@ class ConcurrentHashJoin : public IJoin
     IBlocksStreamPtr
     getNonJoinedBlocks(const Block & left_sample_block, const Block & result_sample_block, UInt64 max_block_size) const override;
 
-
     bool isCloneSupported() const override
     {
         return !getTotals() && getTotalRowCount() == 0;
@@ -88,6 +87,7 @@ class ConcurrentHashJoin : public IJoin
     {
         std::mutex mutex;
         std::unique_ptr<HashJoin> data;
+        bool space_was_preallocated = false;
     };
 
 private:

tests/queries/0_stateless/03319_concurrent_hash_join_double_preallocation_bug.reference

@@ -0,0 +1 @@
+100000

tests/queries/0_stateless/03319_concurrent_hash_join_double_preallocation_bug.sql

@@ -0,0 +1,25 @@
+-- Tags: no-tsan, no-asan, no-msan, no-ubsan, no-parallel-replicas
+
+drop table if exists lhs;
+drop table if exists rhs;
+
+create table lhs(a UInt64, b UInt64) Engine = MergeTree order by tuple();
+create table rhs(a UInt64, b UInt64) Engine = MergeTree order by tuple();
+
+insert into lhs select number, number from numbers_mt(1e5);
+-- rhs should be bigger to trigger tables swap (see `query_plan_join_swap_table`)
+insert into rhs select number, number from numbers_mt(1e6);
+
+set max_threads = 8, query_plan_join_swap_table = 1, join_algorithm = 'parallel_hash', enable_analyzer = 1;
+
+-- First populate the cache of hash table sizes
+select * from lhs as t1 join rhs as t2 on t1.a = t2.a format Null;
+
+-- For the next run we will preallocate the space
+select * from lhs as t1 join rhs as t2 on t1.a = t2.a format Null settings log_comment = '03319_second_query';
+
+system flush logs;
+
+select ProfileEvents['HashJoinPreallocatedElementsInHashTables']
+from system.query_log
+where event_date >= yesterday() and current_database = currentDatabase() and type = 'QueryFinish' and log_comment = '03319_second_query';