Merge pull request ClickHouse#76185 from ClickHouse/parallel_hash_join_threshold

Parallel hash join threshold

Author: nickitat

src/Core/Settings.cpp

@@ -6485,6 +6485,10 @@ As each series represents a node in Keeper, it is recommended to have no more th
 )", 0) \
     DECLARE(Bool, use_hive_partitioning, true, R"(
 When enabled, ClickHouse will detect Hive-style partitioning in path (`/name=value/`) in file-like table engines [File](/sql-reference/table-functions/file#hive-style-partitioning)/[S3](/sql-reference/table-functions/s3#hive-style-partitioning)/[URL](/sql-reference/table-functions/url#hive-style-partitioning)/[HDFS](/sql-reference/table-functions/hdfs#hive-style-partitioning)/[AzureBlobStorage](/sql-reference/table-functions/azureBlobStorage#hive-style-partitioning) and will allow to use partition columns as virtual columns in the query. These virtual columns will have the same names as in the partitioned path, but starting with `_`.
+)", 0) \
+    DECLARE(UInt64, parallel_hash_join_threshold, 100'000, R"(
+When hash-based join algorithm is applied, this threshold helps to decide between using `hash` and `parallel_hash` (only if estimation of the right table size is available).
+The former is used when we know that the right table size is below the threshold.
 )", 0) \
     DECLARE(Bool, apply_settings_from_server, true, R"(
 Whether the client should accept settings from server.

src/Core/SettingsChangesHistory.cpp

@@ -76,6 +76,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
             {"page_cache_block_size", 1048576, 1048576, "Made this setting adjustable on a per-query level."},
             {"page_cache_lookahead_blocks", 16, 16, "Made this setting adjustable on a per-query level."},
             {"output_format_pretty_glue_chunks", "0", "auto", "A new setting to make Pretty formats prettier."},
+            {"parallel_hash_join_threshold", 0, 100'000, "New setting"},
             {"make_distributed_plan", 0, 0, "New experimental setting."},
             {"execute_distributed_plan_locally", 0, 0, "New experimental setting."},
             {"default_shuffle_join_bucket_count", 8, 8, "New experimental setting."},

src/Interpreters/Aggregator.cpp

@@ -1,3 +1,5 @@
+#include <algorithm>
+#include <optional>
 #include <Core/Settings.h>
 #include <Poco/Util/Application.h>
 
@@ -39,10 +41,6 @@
 #include <Common/threadPoolCallbackRunner.h>
 #include <Common/typeid_cast.h>
 
-#include <algorithm>
-#include <numeric>
-#include <optional>
-
 namespace ProfileEvents
 {
     extern const Event ExternalAggregationWritePart;
@@ -140,7 +138,7 @@ void updateStatistics(const DB::ManyAggregatedDataVariants & data_variants, cons
     const auto median_size = sizes.begin() + sizes.size() / 2; // not precisely though...
     std::nth_element(sizes.begin(), median_size, sizes.end());
     const auto sum_of_sizes = std::accumulate(sizes.begin(), sizes.end(), 0ull);
-    DB::getHashTablesStatistics().update(sum_of_sizes, *median_size, params);
+    DB::getHashTablesStatistics<DB::AggregationEntry>().update({.sum_of_sizes = sum_of_sizes, .median_size = *median_size}, params);
 }
 
 DB::ColumnNumbers calculateKeysPositions(const DB::Block & header, const DB::Aggregator::Params & params)

src/Interpreters/ConcurrentHashJoin.cpp

@@ -78,13 +78,17 @@ void updateStatistics(const auto & hash_joins, const DB::StatsCollectingParams &
     if (!params.isCollectionAndUseEnabled())
         return;
 
-    std::vector<size_t> sizes(hash_joins.size());
-    for (size_t i = 0; i < hash_joins.size(); ++i)
-        sizes[i] = hash_joins[i]->data->getTotalRowCount();
-    const auto median_size = sizes.begin() + sizes.size() / 2; // not precisely though...
-    std::nth_element(sizes.begin(), median_size, sizes.end());
-    if (auto sum_of_sizes = std::accumulate(sizes.begin(), sizes.end(), 0ull))
-        DB::getHashTablesStatistics().update(sum_of_sizes, *median_size, params);
+    const auto ht_size = hash_joins.at(0)->data->getTotalRowCount();
+    if (!std::ranges::all_of(hash_joins, [&](const auto & hash_join) { return hash_join->data->getTotalRowCount() == ht_size; }))
+        throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "HashJoin instances have different sizes");
+
+    const auto source_rows = std::accumulate(
+        hash_joins.begin(),
+        hash_joins.end(),
+        0ull,
+        [](auto acc, const auto & hash_join) { return acc + hash_join->data->getJoinedData()->rows_to_join; });
+    if (ht_size)
+        DB::getHashTablesStatistics<DB::HashJoinEntry>().update({.ht_size = ht_size, .source_rows = source_rows}, params);
 }
 
 UInt32 toPowerOfTwo(UInt32 x)
@@ -101,11 +105,11 @@ HashJoin::RightTableDataPtr getData(const std::shared_ptr<ConcurrentHashJoin::In
 
 void reserveSpaceInHashMaps(HashJoin & hash_join, size_t ind, const StatsCollectingParams & stats_collecting_params, size_t slots)
 {
-    if (auto hint = getSizeHint(stats_collecting_params, slots))
+    if (auto hint = getSizeHint(stats_collecting_params))
     {
         /// 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;
+        const size_t reserve_size = hint->ht_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.
@@ -205,7 +209,8 @@ ConcurrentHashJoin::~ConcurrentHashJoin()
         if (!hash_joins[0]->data->twoLevelMapIsUsed())
             return;
 
-        updateStatistics(hash_joins, stats_collecting_params);
+        if (build_phase_finished)
+            updateStatistics(hash_joins, stats_collecting_params);
 
         for (size_t i = 0; i < slots; ++i)
         {
@@ -643,6 +648,8 @@ void ConcurrentHashJoin::onBuildPhaseFinish()
             hash_joins[i]->data->getUsedFlags() = hash_joins[0]->data->getUsedFlags();
         }
     }
+
+    build_phase_finished = true;
 }
 }
 

src/Interpreters/ConcurrentHashJoin.h

@@ -98,6 +98,7 @@ class ConcurrentHashJoin : public IJoin
     bool any_take_last_row;
     std::unique_ptr<ThreadPool> pool;
     std::vector<std::shared_ptr<InternalHashJoin>> hash_joins;
+    bool build_phase_finished = false;
 
     StatsCollectingParams stats_collecting_params;
 

src/Interpreters/HashTablesStatistics.cpp

@@ -11,7 +11,8 @@ namespace ErrorCodes
 extern const int LOGICAL_ERROR;
 }
 
-std::optional<HashTablesStatistics::Entry> HashTablesStatistics::getSizeHint(const Params & params)
+template <typename Entry>
+std::optional<Entry> HashTablesStatistics<Entry>::getSizeHint(const Params & params)
 {
     if (!params.isCollectionAndUseEnabled())
         throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Collection and use of the statistics should be enabled.");
@@ -20,19 +21,15 @@ std::optional<HashTablesStatistics::Entry> HashTablesStatistics::getSizeHint(con
     const auto cache = getHashTableStatsCache(params, lock);
     if (const auto hint = cache->get(params.key))
     {
-        LOG_TRACE(
-            getLogger("HashTablesStatistics"),
-            "An entry for key={} found in cache: sum_of_sizes={}, median_size={}",
-            params.key,
-            hint->sum_of_sizes,
-            hint->median_size);
+        LOG_TRACE(getLogger("HashTablesStatistics"), "An entry for key={} found in cache: {}", params.key, hint->dump());
         return *hint;
     }
     return std::nullopt;
 }
 
 /// Collection and use of the statistics should be enabled.
-void HashTablesStatistics::update(size_t sum_of_sizes, size_t median_size, const Params & params)
+template <typename Entry>
+void HashTablesStatistics<Entry>::update(const Entry & new_entry, const Params & params)
 {
     if (!params.isCollectionAndUseEnabled())
         throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Collection and use of the statistics should be enabled.");
@@ -41,20 +38,15 @@ void HashTablesStatistics::update(size_t sum_of_sizes, size_t median_size, const
     const auto cache = getHashTableStatsCache(params, lock);
     const auto hint = cache->get(params.key);
     // We'll maintain the maximum among all the observed values until another prediction is much lower (that should indicate some change)
-    if (!hint || sum_of_sizes < hint->sum_of_sizes / 2 || hint->sum_of_sizes < sum_of_sizes || median_size < hint->median_size / 2
-        || hint->median_size < median_size)
+    if (!hint || hint->shouldBeUpdated(new_entry))
     {
-        LOG_TRACE(
-            getLogger("HashTablesStatistics"),
-            "Statistics updated for key={}: new sum_of_sizes={}, median_size={}",
-            params.key,
-            sum_of_sizes,
-            median_size);
-        cache->set(params.key, std::make_shared<Entry>(Entry{.sum_of_sizes = sum_of_sizes, .median_size = median_size}));
+        LOG_TRACE(getLogger("HashTablesStatistics"), "Statistics updated for key={}: {}", params.key, new_entry.dump());
+        cache->set(params.key, std::make_shared<Entry>(new_entry));
     }
 }
 
-std::optional<HashTablesCacheStatistics> HashTablesStatistics::getCacheStats() const
+template <typename Entry>
+std::optional<HashTablesCacheStatistics> HashTablesStatistics<Entry>::getCacheStats() const
 {
     std::lock_guard lock(mutex);
     if (hash_table_stats)
@@ -67,29 +59,38 @@ std::optional<HashTablesCacheStatistics> HashTablesStatistics::getCacheStats() c
     return std::nullopt;
 }
 
-HashTablesStatistics::CachePtr HashTablesStatistics::getHashTableStatsCache(const Params & params, const std::lock_guard<std::mutex> &)
+template <typename Entry>
+HashTablesStatistics<Entry>::CachePtr
+HashTablesStatistics<Entry>::getHashTableStatsCache(const Params & params, const std::lock_guard<std::mutex> &)
 {
     if (!hash_table_stats)
         hash_table_stats = std::make_shared<Cache>(params.max_entries_for_hash_table_stats * sizeof(Entry));
     return hash_table_stats;
 }
 
-HashTablesStatistics & getHashTablesStatistics()
-{
-    static HashTablesStatistics hash_tables_stats;
-    return hash_tables_stats;
-}
-
 std::optional<HashTablesCacheStatistics> getHashTablesCacheStatistics()
 {
-    return getHashTablesStatistics().getCacheStats();
+    HashTablesCacheStatistics res{};
+    if (auto aggr_stats = getHashTablesStatistics<AggregationEntry>().getCacheStats())
+    {
+        res.entries += aggr_stats->entries;
+        res.hits += aggr_stats->hits;
+        res.misses += aggr_stats->misses;
+    }
+    if (auto hash_join_stats = getHashTablesStatistics<HashJoinEntry>().getCacheStats())
+    {
+        res.entries += hash_join_stats->entries;
+        res.hits += hash_join_stats->hits;
+        res.misses += hash_join_stats->misses;
+    }
+    return res;
 }
 
-std::optional<HashTablesStatistics::Entry> getSizeHint(const DB::StatsCollectingParams & stats_collecting_params, size_t tables_cnt)
+std::optional<AggregationEntry> getSizeHint(const DB::StatsCollectingParams & stats_collecting_params, size_t tables_cnt)
 {
     if (stats_collecting_params.isCollectionAndUseEnabled())
     {
-        if (auto hint = DB::getHashTablesStatistics().getSizeHint(stats_collecting_params))
+        if (auto hint = DB::getHashTablesStatistics<AggregationEntry>().getSizeHint(stats_collecting_params))
         {
             const auto lower_limit = hint->sum_of_sizes / tables_cnt;
             const auto upper_limit = stats_collecting_params.max_size_to_preallocate / tables_cnt;
@@ -109,10 +110,34 @@ std::optional<HashTablesStatistics::Entry> getSizeHint(const DB::StatsCollecting
             /// https://github.com/ClickHouse/ClickHouse/issues/44402#issuecomment-1359920703
             else if ((tables_cnt > 1 && hint->sum_of_sizes > 100'000) || hint->sum_of_sizes > 500'000)
             {
-                return HashTablesStatistics::Entry{hint->sum_of_sizes, std::max(lower_limit, hint->median_size)};
+                return AggregationEntry{hint->sum_of_sizes, std::max(lower_limit, hint->median_size)};
+            }
+        }
+    }
+    return std::nullopt;
+}
+
+std::optional<HashJoinEntry> getSizeHint(const DB::StatsCollectingParams & stats_collecting_params)
+{
+    if (stats_collecting_params.isCollectionAndUseEnabled())
+    {
+        if (auto hint = DB::getHashTablesStatistics<HashJoinEntry>().getSizeHint(stats_collecting_params))
+        {
+            if (hint->ht_size > stats_collecting_params.max_size_to_preallocate)
+            {
+                LOG_TRACE(
+                    getLogger("HashTablesStatistics"),
+                    "No space were preallocated in hash tables because 'max_size_to_preallocate' has too small value: {}, should be at "
+                    "least {}",
+                    stats_collecting_params.max_size_to_preallocate,
+                    hint->ht_size);
             }
+            return hint;
         }
     }
     return std::nullopt;
 }
+
+template class HashTablesStatistics<AggregationEntry>;
+template class HashTablesStatistics<HashJoinEntry>;
 }

src/Interpreters/HashTablesStatistics.h

@@ -26,22 +26,47 @@ struct StatsCollectingParams
     bool isCollectionAndUseEnabled() const { return key != 0; }
     void disable() { key = 0; }
 
+    StatsCollectingParams & setKey(UInt64 key_)
+    {
+        key = key_;
+        return *this;
+    }
+
     UInt64 key = 0;
     const size_t max_entries_for_hash_table_stats = 0;
     const size_t max_size_to_preallocate = 0;
 };
 
+struct AggregationEntry
+{
+    bool shouldBeUpdated(const AggregationEntry & new_entry) const
+    {
+        return new_entry.sum_of_sizes < sum_of_sizes / 2 || sum_of_sizes < new_entry.sum_of_sizes || new_entry.median_size < median_size / 2
+            || median_size < new_entry.median_size;
+    }
+
+    std::string dump() const { return fmt::format("sum_of_sizes={}, median_size={}", sum_of_sizes, median_size); }
+
+    size_t sum_of_sizes; // used to determine if it's better to convert aggregation to two-level from the beginning
+    size_t median_size; // roughly the size we're going to preallocate on each thread
+};
+
+struct HashJoinEntry
+{
+    bool shouldBeUpdated(const HashJoinEntry & new_entry) const { return new_entry.ht_size < ht_size / 2 || ht_size < new_entry.ht_size; }
+
+    std::string dump() const { return fmt::format("ht_size={}", ht_size); }
+
+    size_t ht_size; // the size of the shared hash table
+    size_t source_rows; // the number of rows in the source table
+};
+
 /** Collects observed HashTable-s sizes to avoid redundant intermediate resizes.
   */
+template <typename Entry>
 class HashTablesStatistics
 {
 public:
-    struct Entry
-    {
-        size_t sum_of_sizes; // used to determine if it's better to convert aggregation to two-level from the beginning
-        size_t median_size; // roughly the size we're going to preallocate on each thread
-    };
-
     using Cache = DB::CacheBase<UInt64, Entry>;
     using CachePtr = std::shared_ptr<Cache>;
     using Params = StatsCollectingParams;
@@ -50,7 +75,7 @@ class HashTablesStatistics
     std::optional<Entry> getSizeHint(const Params & params);
 
     /// Collection and use of the statistics should be enabled.
-    void update(size_t sum_of_sizes, size_t median_size, const Params & params);
+    void update(const Entry & new_entry, const Params & params);
 
     std::optional<DB::HashTablesCacheStatistics> getCacheStats() const;
 
@@ -61,9 +86,15 @@ class HashTablesStatistics
     CachePtr hash_table_stats;
 };
 
-HashTablesStatistics & getHashTablesStatistics();
+template <typename Entry>
+HashTablesStatistics<Entry> & getHashTablesStatistics()
+{
+    static HashTablesStatistics<Entry> hash_tables_stats;
+    return hash_tables_stats;
+}
 
 std::optional<HashTablesCacheStatistics> getHashTablesCacheStatistics();
 
-std::optional<HashTablesStatistics::Entry> getSizeHint(const DB::StatsCollectingParams & stats_collecting_params, size_t tables_cnt);
+std::optional<AggregationEntry> getSizeHint(const DB::StatsCollectingParams & stats_collecting_params, size_t tables_cnt);
+std::optional<HashJoinEntry> getSizeHint(const DB::StatsCollectingParams & stats_collecting_params);
 }

src/Interpreters/JoinInfo.cpp

@@ -39,6 +39,7 @@ namespace Setting
 
     extern const SettingsBool collect_hash_table_stats_during_joins;
     extern const SettingsUInt64 max_size_to_preallocate_for_joins;
+    extern const SettingsUInt64 parallel_hash_join_threshold;
 
     extern const SettingsUInt64 max_joined_block_size_rows;
     extern const SettingsString temporary_files_codec;
@@ -73,6 +74,7 @@ namespace QueryPlanSerializationSetting
 
     extern const QueryPlanSerializationSettingsBool collect_hash_table_stats_during_joins;
     extern const QueryPlanSerializationSettingsUInt64 max_size_to_preallocate_for_joins;
+    extern const QueryPlanSerializationSettingsUInt64 parallel_hash_join_threshold;
 
     extern const QueryPlanSerializationSettingsUInt64 max_joined_block_size_rows;
     extern const QueryPlanSerializationSettingsString temporary_files_codec;
@@ -115,6 +117,7 @@ JoinSettings::JoinSettings(const Settings & query_settings)
 
     collect_hash_table_stats_during_joins = query_settings[Setting::collect_hash_table_stats_during_joins];
     max_size_to_preallocate_for_joins = query_settings[Setting::max_size_to_preallocate_for_joins];
+    parallel_hash_join_threshold = query_settings[Setting::parallel_hash_join_threshold];
 
     temporary_files_codec = query_settings[Setting::temporary_files_codec];
     join_output_by_rowlist_perkey_rows_threshold = query_settings[Setting::join_output_by_rowlist_perkey_rows_threshold];
@@ -148,6 +151,7 @@ JoinSettings::JoinSettings(const QueryPlanSerializationSettings & settings)
 
     collect_hash_table_stats_during_joins = settings[QueryPlanSerializationSetting::collect_hash_table_stats_during_joins];
     max_size_to_preallocate_for_joins = settings[QueryPlanSerializationSetting::max_size_to_preallocate_for_joins];
+    parallel_hash_join_threshold = settings[QueryPlanSerializationSetting::parallel_hash_join_threshold];
 
     max_joined_block_size_rows = settings[QueryPlanSerializationSetting::max_joined_block_size_rows];
     temporary_files_codec = settings[QueryPlanSerializationSetting::temporary_files_codec];
@@ -185,6 +189,7 @@ void JoinSettings::updatePlanSettings(QueryPlanSerializationSettings & settings)
 
     settings[QueryPlanSerializationSetting::collect_hash_table_stats_during_joins] = collect_hash_table_stats_during_joins;
     settings[QueryPlanSerializationSetting::max_size_to_preallocate_for_joins] = max_size_to_preallocate_for_joins;
+    settings[QueryPlanSerializationSetting::parallel_hash_join_threshold] = parallel_hash_join_threshold;
 
     settings[QueryPlanSerializationSetting::max_joined_block_size_rows] = max_joined_block_size_rows;
     settings[QueryPlanSerializationSetting::temporary_files_codec] = temporary_files_codec;

src/Interpreters/JoinInfo.h

@@ -219,6 +219,7 @@ struct JoinSettings
     /* Hash/Parallel hash join settings */
     bool collect_hash_table_stats_during_joins;
     UInt64 max_size_to_preallocate_for_joins;
+    UInt64 parallel_hash_join_threshold;
     UInt64 join_output_by_rowlist_perkey_rows_threshold;
     bool allow_experimental_join_right_table_sorting;
     UInt64 join_to_sort_minimum_perkey_rows;

src/Planner/PlannerJoinTree.cpp

@@ -1515,7 +1515,7 @@ std::tuple<QueryPlan, JoinPtr> buildJoinQueryPlan(
     auto hash_table_stat_cache_key = preCalculateCacheKey(right_table_expression, select_query_info);
     const auto cache_key_for_parallel_hash = calculateCacheKey(table_join, hash_table_stat_cache_key);
     auto join_algorithm = chooseJoinAlgorithm(
-        table_join, prepared_join_storage, left_header, right_header, JoinAlgorithmSettings(*planner_context->getQueryContext()), cache_key_for_parallel_hash);
+        table_join, prepared_join_storage, left_header, right_header, JoinAlgorithmSettings(*planner_context->getQueryContext()), cache_key_for_parallel_hash, /*rhs_size_estimation=*/{});
     auto result_plan = QueryPlan();
 
     bool is_filled_join = join_algorithm->isFilled();