diff --git a/dbms/src/DataStreams/AggregatingBlockInputStream.cpp b/dbms/src/DataStreams/AggregatingBlockInputStream.cpp index 9eb9a761f87..a6bf638a5d2 100644 --- a/dbms/src/DataStreams/AggregatingBlockInputStream.cpp +++ b/dbms/src/DataStreams/AggregatingBlockInputStream.cpp @@ -80,7 +80,8 @@ Block AggregatingBlockInputStream::readImpl() final, 1, 1, - log->identifier()); + log->identifier(), + aggregator.getHashTableStatsProfileInfo()); } } diff --git a/dbms/src/DataStreams/AggregatingBlockInputStream.h b/dbms/src/DataStreams/AggregatingBlockInputStream.h index 6fb3c99e0cd..345fefb7d4a 100644 --- a/dbms/src/DataStreams/AggregatingBlockInputStream.h +++ b/dbms/src/DataStreams/AggregatingBlockInputStream.h @@ -40,7 +40,8 @@ class AggregatingBlockInputStream : public IProfilingBlockInputStream const Aggregator::Params & params_, bool final_, const String & req_id, - const RegisterOperatorSpillContext & register_operator_spill_context) + const RegisterOperatorSpillContext & register_operator_spill_context, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info = nullptr) : log(Logger::get(req_id)) , params(params_) , aggregator( @@ -49,7 +50,8 @@ class AggregatingBlockInputStream : public IProfilingBlockInputStream 1, register_operator_spill_context, /*is_auto_pass_through=*/false, - params.use_magic_hash) + params.use_magic_hash, + hash_table_stats_profile_info) , final(final_) { children.push_back(input); diff --git a/dbms/src/DataStreams/AutoPassThroughAggregatingBlockInputStream.h b/dbms/src/DataStreams/AutoPassThroughAggregatingBlockInputStream.h index b0dad20e881..d3bb061a648 100644 --- a/dbms/src/DataStreams/AutoPassThroughAggregatingBlockInputStream.h +++ b/dbms/src/DataStreams/AutoPassThroughAggregatingBlockInputStream.h @@ -32,7 +32,8 @@ class AutoPassThroughAggregatingBlockInputStream : public IProfilingBlockInputSt const BlockInputStreamPtr & input_, const Aggregator::Params & params_, const String & req_id, - UInt64 row_limit_unit) + UInt64 row_limit_unit, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info = nullptr) { children.push_back(input_); auto_pass_through_context = std::make_unique( @@ -40,7 +41,10 @@ class AutoPassThroughAggregatingBlockInputStream : public IProfilingBlockInputSt params_, [&]() { return this->isCancelled(); }, req_id, - row_limit_unit); + row_limit_unit, + AutoPassThroughHashAggContext::DEF_NORMAL_UNIT_NUM, + AutoPassThroughHashAggContext::DEF_DYNAMIC_UNIT_NUM, + hash_table_stats_profile_info); } String getName() const override { return NAME; } diff --git a/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.cpp b/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.cpp index 25cf6499c80..39b765fa182 100644 --- a/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.cpp +++ b/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.cpp @@ -83,7 +83,8 @@ MergingAggregatedMemoryEfficientBlockInputStream::MergingAggregatedMemoryEfficie bool final_, size_t reading_threads_, size_t merging_threads_, - const String & req_id) + const String & req_id, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info) : log(Logger::get(req_id)) , aggregator( params, @@ -91,7 +92,8 @@ MergingAggregatedMemoryEfficientBlockInputStream::MergingAggregatedMemoryEfficie merging_threads_, [](const OperatorSpillContextPtr &) {}, /*is_auto_pass_through=*/false, - params.use_magic_hash) + params.use_magic_hash, + hash_table_stats_profile_info) , final(final_) , reading_threads(std::min(reading_threads_, inputs_.size())) , merging_threads(merging_threads_) diff --git a/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.h b/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.h index 02cf5fab23f..41671907c7a 100644 --- a/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.h +++ b/dbms/src/DataStreams/MergingAggregatedMemoryEfficientBlockInputStream.h @@ -79,7 +79,8 @@ class MergingAggregatedMemoryEfficientBlockInputStream final : public IProfiling bool final_, size_t reading_threads_, size_t merging_threads_, - const String & req_id); + const String & req_id, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info = nullptr); ~MergingAggregatedMemoryEfficientBlockInputStream() override; diff --git a/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.cpp b/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.cpp index b32b96da15e..d07689f6e79 100644 --- a/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.cpp +++ b/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.cpp @@ -34,7 +34,8 @@ ParallelAggregatingBlockInputStream::ParallelAggregatingBlockInputStream( Int64 max_buffered_bytes_, size_t temporary_data_merge_threads_, const String & req_id, - const RegisterOperatorSpillContext & register_operator_spill_context) + const RegisterOperatorSpillContext & register_operator_spill_context, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info) : log(Logger::get(req_id)) , max_threads(std::min(inputs.size(), max_threads_)) , params(params_) @@ -44,7 +45,8 @@ ParallelAggregatingBlockInputStream::ParallelAggregatingBlockInputStream( max_threads, register_operator_spill_context, /*is_auto_pass_through=*/false, - params.use_magic_hash) + params.use_magic_hash, + hash_table_stats_profile_info) , final(final_) , max_buffered_bytes(max_buffered_bytes_) , temporary_data_merge_threads(temporary_data_merge_threads_) @@ -137,7 +139,8 @@ Block ParallelAggregatingBlockInputStream::readImpl() final, temporary_data_merge_threads, temporary_data_merge_threads, - log->identifier()); + log->identifier(), + aggregator.getHashTableStatsProfileInfo()); } executed = true; diff --git a/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.h b/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.h index fdbb6107b07..b3135bd419d 100644 --- a/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.h +++ b/dbms/src/DataStreams/ParallelAggregatingBlockInputStream.h @@ -42,7 +42,8 @@ class ParallelAggregatingBlockInputStream : public IProfilingBlockInputStream Int64 max_buffered_bytes_, size_t temporary_data_merge_threads_, const String & req_id, - const RegisterOperatorSpillContext & register_operator_spill_context); + const RegisterOperatorSpillContext & register_operator_spill_context, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info = nullptr); String getName() const override { return NAME; } diff --git a/dbms/src/Flash/Coprocessor/DAGContext.cpp b/dbms/src/Flash/Coprocessor/DAGContext.cpp index 84b81c383f2..6211dc4c6d4 100644 --- a/dbms/src/Flash/Coprocessor/DAGContext.cpp +++ b/dbms/src/Flash/Coprocessor/DAGContext.cpp @@ -355,6 +355,21 @@ std::unordered_map & DAGContext::getJoinExecuteInfoMap( return join_execute_info_map; } +void DAGContext::addAggregationProfileInfo(const String & executor_id, const AggregationProfileInfoPtr & profile_info) +{ + RUNTIME_CHECK(profile_info); + std::lock_guard lock(operator_profile_infos_map_mu); + const auto [it, inserted] = aggregation_profile_info_map.emplace(executor_id, profile_info); + RUNTIME_CHECK(inserted || it->second == profile_info); +} + +AggregationProfileInfoPtr DAGContext::getAggregationProfileInfo(const String & executor_id) +{ + std::lock_guard lock(operator_profile_infos_map_mu); + const auto it = aggregation_profile_info_map.find(executor_id); + return it == aggregation_profile_info_map.end() ? nullptr : it->second; +} + std::unordered_map & DAGContext::getInBoundIOInputStreamsMap() { return inbound_io_input_streams_map; diff --git a/dbms/src/Flash/Coprocessor/DAGContext.h b/dbms/src/Flash/Coprocessor/DAGContext.h index 873fbc5f628..e8a8ae5e678 100644 --- a/dbms/src/Flash/Coprocessor/DAGContext.h +++ b/dbms/src/Flash/Coprocessor/DAGContext.h @@ -100,6 +100,7 @@ struct JoinProfileInfo mutable std::mutex hash_table_stats_mutex; }; using JoinProfileInfoPtr = std::shared_ptr; +using AggregationProfileInfoPtr = HashTableStatsProfileInfoPtr; struct JoinExecuteInfo { String build_side_root_executor_id; @@ -238,6 +239,10 @@ class DAGContext std::unordered_map & getJoinExecuteInfoMap(); + void addAggregationProfileInfo(const String & executor_id, const AggregationProfileInfoPtr & profile_info); + + AggregationProfileInfoPtr getAggregationProfileInfo(const String & executor_id); + std::unordered_map & getInBoundIOInputStreamsMap(); std::unordered_map & getInboundIOProfileInfosMap(); @@ -549,6 +554,8 @@ class DAGContext /// join_execute_info_map is a map that maps from join_probe_executor_id to JoinExecuteInfo /// DAGResponseWriter / JoinStatistics gets JoinExecuteInfo through it. std::unordered_map join_execute_info_map; + /// aggregation_profile_info_map maps an aggregation executor to its hash-table statistics accumulator. + std::unordered_map aggregation_profile_info_map; /// inbound_io_input_streams_map is a map that maps from executor_id (table_scan / exchange_receiver) to BlockInputStreams. /// BlockInputStreams contains ExchangeReceiverInputStream, CoprocessorBlockInputStream and local_read_input_stream etc. std::unordered_map inbound_io_input_streams_map; diff --git a/dbms/src/Flash/Coprocessor/HashTableStats.h b/dbms/src/Flash/Coprocessor/HashTableStats.h index 0f1580a8e08..ee8a1e8b893 100644 --- a/dbms/src/Flash/Coprocessor/HashTableStats.h +++ b/dbms/src/Flash/Coprocessor/HashTableStats.h @@ -17,6 +17,10 @@ #include #include +#include +#include +#include + namespace DB { enum class HashTableSizeKind : UInt8 @@ -41,4 +45,29 @@ struct HashTableStats memory_bytes += other.memory_bytes; } }; + +/// Thread-safe accumulator shared by all runtime fragments of one physical hash-table operator. +class HashTableStatsProfileInfo +{ +public: + void mergeHashTableStats(const HashTableStats & stats) + { + std::lock_guard lock(hash_table_stats_mutex); + if (!hash_table_stats) + hash_table_stats = stats; + else + hash_table_stats->merge(stats); + } + + std::optional getHashTableStats() const + { + std::lock_guard lock(hash_table_stats_mutex); + return hash_table_stats; + } + +private: + mutable std::mutex hash_table_stats_mutex; + std::optional hash_table_stats; +}; +using HashTableStatsProfileInfoPtr = std::shared_ptr; } // namespace DB diff --git a/dbms/src/Flash/Planner/Plans/PhysicalAggregation.cpp b/dbms/src/Flash/Planner/Plans/PhysicalAggregation.cpp index fda79f964cc..b5a3f6dd5fd 100644 --- a/dbms/src/Flash/Planner/Plans/PhysicalAggregation.cpp +++ b/dbms/src/Flash/Planner/Plans/PhysicalAggregation.cpp @@ -35,6 +35,18 @@ namespace DB { +HashTableStatsProfileInfoPtr PhysicalAggregation::initHashTableStatsProfileInfo(Context & context) +{ + if (!hash_table_stats_profile_info) + { + hash_table_stats_profile_info = std::make_shared(); + if (auto * dag_context = context.getDAGContext(); dag_context != nullptr) + dag_context->addAggregationProfileInfo(executor_id, hash_table_stats_profile_info); + } + + return hash_table_stats_profile_info; +} + PhysicalPlanNodePtr PhysicalAggregation::build( const Context & context, const String & executor_id, @@ -121,6 +133,8 @@ void PhysicalAggregation::buildBlockInputStreamImpl(DAGPipeline & pipeline, Cont // Fine grained shuffle is for 2nd agg, auto pass through is for 1st agg. RUNTIME_CHECK(!(fine_grained_shuffle.enabled() && auto_pass_through_switcher.enabled())); + const auto hash_table_stats_profile = initHashTableStatsProfileInfo(context); + child->buildBlockInputStream(pipeline, context, max_streams); executeExpression(pipeline, before_agg_actions, log, "before aggregation"); @@ -167,7 +181,8 @@ void PhysicalAggregation::buildBlockInputStreamImpl(DAGPipeline & pipeline, Cont fine_grained_spill_context->addOperatorSpillContext(operator_spill_context); else if (context.getDAGContext() != nullptr) context.getDAGContext()->registerOperatorSpillContext(operator_spill_context); - }); + }, + hash_table_stats_profile); stream->setExtraInfo(String(enableFineGrainedShuffleExtraInfo)); }); if (fine_grained_spill_context != nullptr) @@ -182,7 +197,8 @@ void PhysicalAggregation::buildBlockInputStreamImpl(DAGPipeline & pipeline, Cont stream, params, log->identifier(), - context.getSettings().max_block_size); + context.getSettings().max_block_size, + hash_table_stats_profile); stream->setExtraInfo(String(autoPassThroughAggregatingExtraInfo)); }); } @@ -193,7 +209,8 @@ void PhysicalAggregation::buildBlockInputStreamImpl(DAGPipeline & pipeline, Cont stream, params, log->identifier(), - context.getSettings().max_block_size); + context.getSettings().max_block_size, + hash_table_stats_profile); stream->setExtraInfo(String(autoPassThroughAggregatingExtraInfo)); }); } @@ -222,7 +239,8 @@ void PhysicalAggregation::buildBlockInputStreamImpl(DAGPipeline & pipeline, Cont { context.getDAGContext()->registerOperatorSpillContext(operator_spill_context); } - }); + }, + hash_table_stats_profile); pipeline.streams.resize(1); pipeline.firstStream() = std::move(stream); @@ -246,7 +264,8 @@ void PhysicalAggregation::buildBlockInputStreamImpl(DAGPipeline & pipeline, Cont { context.getDAGContext()->registerOperatorSpillContext(operator_spill_context); } - }); + }, + hash_table_stats_profile); } // we can record for agg after restore concurrency. @@ -266,6 +285,7 @@ void PhysicalAggregation::buildPipelineExecGroupImpl( // Because for non fine grained shuffle, AggregateBuild and AggregateConvergent will be used to build aggregation. // Also auto pass through hashagg use PhysicalAggregation to build. But they cannot be true at the same time. RUNTIME_CHECK(fine_grained_shuffle.enabled() != auto_pass_through_switcher.enabled()); + const auto hash_table_stats_profile = initHashTableStatsProfileInfo(context); // Auto pass through hashagg doesn't handle empty_result_for_aggregation_by_empty_set. // Also tidb shouldn't generate this kind plan because all data is aggregated into one row if keys_size == 0. @@ -308,7 +328,8 @@ void PhysicalAggregation::buildPipelineExecGroupImpl( exec_context, log->identifier(), params, - fine_grained_spill_context)); + fine_grained_spill_context, + hash_table_stats_profile)); }); if (fine_grained_spill_context != nullptr) context.getDAGContext()->registerOperatorSpillContext(fine_grained_spill_context); @@ -323,7 +344,8 @@ void PhysicalAggregation::buildPipelineExecGroupImpl( exec_context, params, log->identifier(), - context.getSettings().max_block_size)); + context.getSettings().max_block_size, + hash_table_stats_profile)); }); } else if (auto_pass_through_switcher.isAuto()) @@ -334,7 +356,8 @@ void PhysicalAggregation::buildPipelineExecGroupImpl( exec_context, params, log->identifier(), - context.getSettings().max_block_size)); + context.getSettings().max_block_size, + hash_table_stats_profile)); }); } else @@ -358,7 +381,8 @@ void PhysicalAggregation::buildPipeline( Context & context, PipelineExecutorContext & exec_context) { - auto aggregate_context = std::make_shared(log->identifier()); + const auto hash_table_stats_profile = initHashTableStatsProfileInfo(context); + auto aggregate_context = std::make_shared(log->identifier(), hash_table_stats_profile); // fine_grained_shuffle and auto_pass_through cannot be ture at the same time. RUNTIME_CHECK(!(fine_grained_shuffle.enabled() && auto_pass_through_switcher.enabled())); if (fine_grained_shuffle.enabled() || auto_pass_through_switcher.enabled()) diff --git a/dbms/src/Flash/Planner/Plans/PhysicalAggregation.h b/dbms/src/Flash/Planner/Plans/PhysicalAggregation.h index 6be21ded21b..a6384a70363 100644 --- a/dbms/src/Flash/Planner/Plans/PhysicalAggregation.h +++ b/dbms/src/Flash/Planner/Plans/PhysicalAggregation.h @@ -14,6 +14,7 @@ #pragma once +#include #include #include #include @@ -67,6 +68,8 @@ class PhysicalAggregation : public PhysicalUnary const Block & getSampleBlock() const override; private: + HashTableStatsProfileInfoPtr initHashTableStatsProfileInfo(Context & context); + void buildBlockInputStreamImpl(DAGPipeline & pipeline, Context & context, size_t max_streams) override; void buildPipelineExecGroupImpl( @@ -85,5 +88,6 @@ class PhysicalAggregation : public PhysicalUnary const AutoPassThroughSwitcher auto_pass_through_switcher; AggregateDescriptions aggregate_descriptions; ExpressionActionsPtr expr_after_agg; + HashTableStatsProfileInfoPtr hash_table_stats_profile_info; }; } // namespace DB diff --git a/dbms/src/Flash/Statistics/CommonExecutorImpl.h b/dbms/src/Flash/Statistics/CommonExecutorImpl.h index 91adf116a92..8da078ac4a1 100644 --- a/dbms/src/Flash/Statistics/CommonExecutorImpl.h +++ b/dbms/src/Flash/Statistics/CommonExecutorImpl.h @@ -14,6 +14,7 @@ #pragma once +#include #include #include @@ -29,7 +30,28 @@ struct AggImpl static bool isSourceExecutor() { return false; } }; -using AggStatistics = ExecutorStatistics; +using AggStatisticsBase = ExecutorStatistics; + +class AggStatistics : public AggStatisticsBase +{ +public: + AggStatistics(const tipb::Executor * executor, DAGContext & dag_context) + : AggStatisticsBase(executor, dag_context) + {} + + void fillExtraExecutionSummary(ExecutionSummary & summary) const override + { + // The statistics collector for an MPP task is initialized before the physical + // plan is built. The aggregation profile is registered while building that + // plan, so it must be looked up when the summary is filled rather than cached + // in the constructor above. + const auto hash_table_stats_profile_info = dag_context.getAggregationProfileInfo(executor_id); + if (!hash_table_stats_profile_info) + return; + if (const auto stats = hash_table_stats_profile_info->getHashTableStats(); stats) + summary.hash_table_stats = *stats; + } +}; struct WindowImpl { diff --git a/dbms/src/Flash/tests/gtest_execution_summary.cpp b/dbms/src/Flash/tests/gtest_execution_summary.cpp index 61364cd3065..a3ab3f942dc 100644 --- a/dbms/src/Flash/tests/gtest_execution_summary.cpp +++ b/dbms/src/Flash/tests/gtest_execution_summary.cpp @@ -148,6 +148,59 @@ class ExecutionSummaryTestRunner : public DB::tests::ExecutorTest } } + void testHashAggTableStats(bool enable_pipeline, bool force_two_level) + { + enablePipeline(enable_pipeline); + const auto old_two_level_threshold = context.context->getSettingsRef().group_by_two_level_threshold; + const auto old_two_level_threshold_bytes = context.context->getSettingsRef().group_by_two_level_threshold_bytes; + if (force_two_level) + { + context.context->setSetting("group_by_two_level_threshold", Field(static_cast(1))); + context.context->setSetting("group_by_two_level_threshold_bytes", Field(static_cast(1))); + } + + auto request = context.scan("test_db", "test_table").aggregation({col("s2")}, {col("s2")}).build(context); + request->set_collect_execution_summaries(true); + + DAGContext dag_context(*request, "test_execution_summary", concurrency); + ExecutorStatisticsCollector statistics_collector("test_execution_summary", true); + // MPP initializes the collector before queryExecute builds the physical plan. + // Keep the same ordering here to verify that aggregation profile lookup is + // deferred until execution summaries are generated. + statistics_collector.initialize(&dag_context); + executeStreams(&dag_context); + statistics_collector.setLocalRUConsumption( + RUConsumption{.cpu_ru = 0.0, .cpu_time_ns = 0, .read_ru = 0.0, .read_bytes = 0}); + const auto summaries = statistics_collector.genExecutionSummaryResponse().execution_summaries(); + + // Restore the test settings before validating the result so an assertion failure + // cannot leak the forced two-level thresholds into subsequent test cases. + context.context->setSetting( + "group_by_two_level_threshold", + Field(static_cast(old_two_level_threshold))); + context.context->setSetting( + "group_by_two_level_threshold_bytes", + Field(static_cast(old_two_level_threshold_bytes))); + + const tipb::ExecutorExecutionSummary * aggregation_summary = nullptr; + for (const auto & summary : summaries) + { + if (summary.has_executor_id() && summary.executor_id() == "aggregation_1") + { + aggregation_summary = &summary; + break; + } + } + ASSERT_NE(aggregation_summary, nullptr); + ASSERT_TRUE(aggregation_summary->has_tiflash_hash_table_stats()); + ASSERT_EQ( + aggregation_summary->tiflash_hash_table_stats().size_kind(), + tipb::TIFLASH_HASH_TABLE_SIZE_KIND_DISTINCT_KEY_COUNT); + // Ten build workers own independent maps, so the reported total is not the final merged NDV. + ASSERT_GE(aggregation_summary->tiflash_hash_table_stats().size(), 3); + ASSERT_GT(aggregation_summary->tiflash_hash_table_stats().memory_bytes(), 0); + } + #define WRAP_FOR_EXCUTION_SUMMARY_TEST_BEGIN \ std::vector type{DAGRequestType::tree, DAGRequestType::list}; \ std::vector pipeline_bools{false, true}; \ @@ -259,6 +312,15 @@ try } CATCH +TEST_F(ExecutionSummaryTestRunner, hashAggTableStats) +try +{ + testHashAggTableStats(false, false); + testHashAggTableStats(true, false); + testHashAggTableStats(true, true); +} +CATCH + TEST_F(ExecutionSummaryTestRunner, genMPPTaskExecutionInfoWithoutSetRUInfo) try { diff --git a/dbms/src/Interpreters/Aggregator.cpp b/dbms/src/Interpreters/Aggregator.cpp index e34e93b2e06..2267326fa43 100644 --- a/dbms/src/Interpreters/Aggregator.cpp +++ b/dbms/src/Interpreters/Aggregator.cpp @@ -296,8 +296,10 @@ Aggregator::Aggregator( size_t concurrency, const RegisterOperatorSpillContext & register_operator_spill_context, bool is_auto_pass_through_, - bool use_magic_hash_) + bool use_magic_hash_, + HashTableStatsProfileInfoPtr hash_table_stats_profile_info_) : params(params_) + , hash_table_stats_profile_info(std::move(hash_table_stats_profile_info_)) , log(Logger::get(req_id)) , is_cancelled([]() { return false; }) , is_auto_pass_through(is_auto_pass_through_) @@ -2435,6 +2437,39 @@ void NO_INLINE Aggregator::mergeBucketImpl(ManyAggregatedDataVariants & data, In } } +void Aggregator::reportHashTableStats(const ManyAggregatedDataVariants & data_variants) const +{ + if (!hash_table_stats_profile_info || params.keys_size == 0) + return; + + HashTableStats stats{ + .size_kind = HashTableSizeKind::DistinctKeyCount, + }; + bool has_hash_table = false; + for (const auto & data : data_variants) + { + if (!data || !data->inited()) + continue; + + has_hash_table = true; + stats.size += data->size(); + stats.memory_bytes += data->bytesCount(); + } + if (has_hash_table) + hash_table_stats_profile_info->mergeHashTableStats(stats); +} + +void Aggregator::reportHashTableStats(const AggregatedDataVariants & data) const +{ + if (!hash_table_stats_profile_info || params.keys_size == 0 || !data.inited()) + return; + + hash_table_stats_profile_info->mergeHashTableStats({ + .size = data.size(), + .size_kind = HashTableSizeKind::DistinctKeyCount, + .memory_bytes = data.bytesCount(), + }); +} MergingBucketsPtr Aggregator::mergeAndConvertToBlocks( ManyAggregatedDataVariants & data_variants, @@ -2444,6 +2479,12 @@ MergingBucketsPtr Aggregator::mergeAndConvertToBlocks( if (unlikely(data_variants.empty())) throw Exception("Empty data passed to Aggregator::mergeAndConvertToBlocks.", ErrorCodes::EMPTY_DATA_PASSED); + if (!hash_table_stats_reported) + { + reportHashTableStats(data_variants); + hash_table_stats_reported = true; + } + LOG_TRACE(log, "Merging aggregated data"); ManyAggregatedDataVariants non_empty_data; @@ -2702,6 +2743,9 @@ BlocksList Aggregator::vstackBlocks(BlocksList & blocks, bool final) #undef M } + // Each restore bucket builds one final effective hash table. Report it before aggregate states move out. + reportHashTableStats(result); + BlocksList return_blocks; if (result.type == AggregatedDataVariants::Type::without_key) return_blocks = prepareBlocksAndFillWithoutKey(result, final); diff --git a/dbms/src/Interpreters/Aggregator.h b/dbms/src/Interpreters/Aggregator.h index 415eef0caec..8e6248de5fc 100644 --- a/dbms/src/Interpreters/Aggregator.h +++ b/dbms/src/Interpreters/Aggregator.h @@ -31,6 +31,7 @@ #include #include #include +#include #include #include #include @@ -947,7 +948,8 @@ class Aggregator size_t concurrency, const RegisterOperatorSpillContext & register_operator_spill_context, bool is_auto_pass_through_, - bool use_magic_hash_); + bool use_magic_hash_, + HashTableStatsProfileInfoPtr hash_table_stats_profile_info_ = nullptr); /// Aggregate the source. Get the result in the form of one of the data structures. void execute(const BlockInputStreamPtr & stream, AggregatedDataVariants & result, size_t thread_num); @@ -1067,15 +1069,22 @@ class Aggregator Block getSourceHeader() const; const Params & getParams() const { return params; } + const HashTableStatsProfileInfoPtr & getHashTableStatsProfileInfo() const { return hash_table_stats_profile_info; } protected: friend struct AggregatedDataVariants; friend class MergingBuckets; + void reportHashTableStats(const ManyAggregatedDataVariants & data_variants) const; + void reportHashTableStats(const AggregatedDataVariants & data) const; + Params params; AggregatedDataVariants::Type method_chosen; + HashTableStatsProfileInfoPtr hash_table_stats_profile_info; + mutable bool hash_table_stats_reported = false; + Sizes key_sizes; diff --git a/dbms/src/Operators/AggregateContext.cpp b/dbms/src/Operators/AggregateContext.cpp index 7bc2cd525ed..60d52115e21 100644 --- a/dbms/src/Operators/AggregateContext.cpp +++ b/dbms/src/Operators/AggregateContext.cpp @@ -33,7 +33,8 @@ void AggregateContext::initBuild( max_threads, register_operator_spill_context, /*is_auto_pass_through=*/false, - params.use_magic_hash); + params.use_magic_hash, + hash_table_stats_profile_info); aggregator->setCancellationHook(is_cancelled); aggregator->initThresholdByAggregatedDataVariantsSize(max_threads); many_data.reserve(max_threads); diff --git a/dbms/src/Operators/AggregateContext.h b/dbms/src/Operators/AggregateContext.h index cb1a6ad20d4..104c48f2119 100644 --- a/dbms/src/Operators/AggregateContext.h +++ b/dbms/src/Operators/AggregateContext.h @@ -37,8 +37,11 @@ struct ThreadData class AggregateContext { public: - explicit AggregateContext(const String & req_id) - : log(Logger::get(req_id)) + explicit AggregateContext( + const String & req_id, + HashTableStatsProfileInfoPtr hash_table_stats_profile_info_ = nullptr) + : hash_table_stats_profile_info(std::move(hash_table_stats_profile_info_)) + , log(Logger::get(req_id)) {} void initBuild( @@ -94,6 +97,7 @@ class AggregateContext private: std::unique_ptr aggregator; + HashTableStatsProfileInfoPtr hash_table_stats_profile_info; size_t keys_size = 0; bool empty_result_for_aggregation_by_empty_set = false; diff --git a/dbms/src/Operators/AutoPassThroughAggregateTransform.h b/dbms/src/Operators/AutoPassThroughAggregateTransform.h index 015ba4a57dd..0fa743cc756 100644 --- a/dbms/src/Operators/AutoPassThroughAggregateTransform.h +++ b/dbms/src/Operators/AutoPassThroughAggregateTransform.h @@ -29,7 +29,8 @@ class AutoPassThroughAggregateTransform : public TransformOp PipelineExecutorContext & exec_context_, const Aggregator::Params & params_, const String & req_id_, - UInt64 row_limit_unit) + UInt64 row_limit_unit, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info = nullptr) : TransformOp(exec_context_, req_id_) , status(Status::building_hash_map) { @@ -38,7 +39,10 @@ class AutoPassThroughAggregateTransform : public TransformOp params_, [&]() { return exec_context.isCancelled(); }, req_id_, - row_limit_unit); + row_limit_unit, + AutoPassThroughHashAggContext::DEF_NORMAL_UNIT_NUM, + AutoPassThroughHashAggContext::DEF_DYNAMIC_UNIT_NUM, + hash_table_stats_profile_info); } String getName() const override { return "AutoPassThroughAggregateTransform"; } diff --git a/dbms/src/Operators/AutoPassThroughHashAggContext.h b/dbms/src/Operators/AutoPassThroughHashAggContext.h index 9ffd48ed8bc..9bfaa96b474 100644 --- a/dbms/src/Operators/AutoPassThroughHashAggContext.h +++ b/dbms/src/Operators/AutoPassThroughHashAggContext.h @@ -50,6 +50,9 @@ struct AutoPassThroughSwitcher class AutoPassThroughHashAggContext { public: + static constexpr size_t DEF_NORMAL_UNIT_NUM = 1; + static constexpr size_t DEF_DYNAMIC_UNIT_NUM = 5; + AutoPassThroughHashAggContext( const Block & child_header_, const Aggregator::Params & params_, @@ -57,7 +60,8 @@ class AutoPassThroughHashAggContext const String & req_id_, UInt64 row_limit_unit_, UInt64 normal_unit_num_ = DEF_NORMAL_UNIT_NUM, - UInt64 dynamic_unit_num_ = DEF_DYNAMIC_UNIT_NUM) + UInt64 dynamic_unit_num_ = DEF_DYNAMIC_UNIT_NUM, + HashTableStatsProfileInfoPtr hash_table_stats_profile_info_ = nullptr) : state(State::Init) , many_data(std::vector(1, nullptr)) , normal_row_limit(row_limit_unit_ * normal_unit_num_) @@ -72,7 +76,8 @@ class AutoPassThroughHashAggContext /*concurrency=*/1, nullptr, /*is_auto_pass_through=*/true, - params_.use_magic_hash); + params_.use_magic_hash, + std::move(hash_table_stats_profile_info_)); aggregator->setCancellationHook(hook); aggregator->initThresholdByAggregatedDataVariantsSize(1); RUNTIME_CHECK(aggregator->getParams().keys_size > 0); @@ -230,9 +235,6 @@ class AutoPassThroughHashAggContext static constexpr size_t INIT_STATE_HASHMAP_THRESHOLD = 2 * 1024 * 1024; static constexpr size_t MAX_DYNAMIC_UNIT_LIMIT = 100; - static constexpr size_t DEF_NORMAL_UNIT_NUM = 1; - static constexpr size_t DEF_DYNAMIC_UNIT_NUM = 5; - std::vector column_generators; }; diff --git a/dbms/src/Operators/LocalAggregateTransform.cpp b/dbms/src/Operators/LocalAggregateTransform.cpp index a105e62406c..6774e28ea09 100644 --- a/dbms/src/Operators/LocalAggregateTransform.cpp +++ b/dbms/src/Operators/LocalAggregateTransform.cpp @@ -32,10 +32,11 @@ LocalAggregateTransform::LocalAggregateTransform( PipelineExecutorContext & exec_context_, const String & req_id, const Aggregator::Params & params_, - const std::shared_ptr & fine_grained_spill_context) + const std::shared_ptr & fine_grained_spill_context, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info) : TransformOp(exec_context_, req_id) , params(params_) - , agg_context(req_id) + , agg_context(req_id, hash_table_stats_profile_info) { agg_context.initBuild( params, diff --git a/dbms/src/Operators/LocalAggregateTransform.h b/dbms/src/Operators/LocalAggregateTransform.h index 235ab28c682..b3af96d0d61 100644 --- a/dbms/src/Operators/LocalAggregateTransform.h +++ b/dbms/src/Operators/LocalAggregateTransform.h @@ -28,7 +28,8 @@ class LocalAggregateTransform : public TransformOp PipelineExecutorContext & exec_context_, const String & req_id, const Aggregator::Params & params_, - const std::shared_ptr & fine_grained_spill_context); + const std::shared_ptr & fine_grained_spill_context, + const HashTableStatsProfileInfoPtr & hash_table_stats_profile_info = nullptr); String getName() const override { return "LocalAggregateTransform"; }