mirror of
https://github.com/ClickHouse/ClickHouse.git
synced 2024-11-24 00:22:29 +00:00
Merge pull request #71898 from ClickHouse/fix-benchmark
Fix `clickhouse-benchmark` metrics report
This commit is contained in:
Alexey Milovidov
GitHub
commit
7007ce7596
@ -7,7 +7,6 @@
|
||||
#include <random>
|
||||
#include <string_view>
|
||||
#include <pcg_random.hpp>
|
||||
#include <Poco/UUID.h>
|
||||
#include <Poco/UUIDGenerator.h>
|
||||
#include <Poco/Util/Application.h>
|
||||
#include <Common/Stopwatch.h>
|
||||
@ -152,8 +151,6 @@ public:
|
||||
global_context->setClientName(std::string(DEFAULT_CLIENT_NAME));
|
||||
global_context->setQueryKindInitial();
|
||||
|
||||
std::cerr << std::fixed << std::setprecision(3);
|
||||
|
||||
/// This is needed to receive blocks with columns of AggregateFunction data type
|
||||
/// (example: when using stage = 'with_mergeable_state')
|
||||
registerAggregateFunctions();
|
||||
@ -226,6 +223,8 @@ private:
|
||||
ContextMutablePtr global_context;
|
||||
QueryProcessingStage::Enum query_processing_stage;
|
||||
|
||||
WriteBufferFromFileDescriptor log{STDERR_FILENO};
|
||||
|
||||
std::atomic<size_t> consecutive_errors{0};
|
||||
|
||||
/// Don't execute new queries after timelimit or SIGINT or exception
|
||||
@ -303,16 +302,16 @@ private:
|
||||
}
|
||||
|
||||
|
||||
std::cerr << "Loaded " << queries.size() << " queries.\n";
|
||||
log << "Loaded " << queries.size() << " queries.\n" << flush;
|
||||
}
|
||||
|
||||
|
||||
void printNumberOfQueriesExecuted(size_t num)
|
||||
{
|
||||
std::cerr << "\nQueries executed: " << num;
|
||||
log << "\nQueries executed: " << num;
|
||||
if (queries.size() > 1)
|
||||
std::cerr << " (" << (num * 100.0 / queries.size()) << "%)";
|
||||
std::cerr << ".\n";
|
||||
log << " (" << (num * 100.0 / queries.size()) << "%)";
|
||||
log << ".\n" << flush;
|
||||
}
|
||||
|
||||
/// Try push new query and check cancellation conditions
|
||||
@ -339,19 +338,19 @@ private:
|
||||
|
||||
if (interrupt_listener.check())
|
||||
{
|
||||
std::cout << "Stopping launch of queries. SIGINT received." << std::endl;
|
||||
std::cout << "Stopping launch of queries. SIGINT received.\n";
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
double seconds = delay_watch.elapsedSeconds();
|
||||
if (delay > 0 && seconds > delay)
|
||||
{
|
||||
printNumberOfQueriesExecuted(queries_executed);
|
||||
cumulative
|
||||
? report(comparison_info_total, total_watch.elapsedSeconds())
|
||||
: report(comparison_info_per_interval, seconds);
|
||||
delay_watch.restart();
|
||||
}
|
||||
double seconds = delay_watch.elapsedSeconds();
|
||||
if (delay > 0 && seconds > delay)
|
||||
{
|
||||
printNumberOfQueriesExecuted(queries_executed);
|
||||
cumulative
|
||||
? report(comparison_info_total, total_watch.elapsedSeconds())
|
||||
: report(comparison_info_per_interval, seconds);
|
||||
delay_watch.restart();
|
||||
}
|
||||
|
||||
return true;
|
||||
@ -438,16 +437,16 @@ private:
|
||||
catch (...)
|
||||
{
|
||||
std::lock_guard lock(mutex);
|
||||
std::cerr << "An error occurred while processing the query " << "'" << query << "'"
|
||||
<< ": " << getCurrentExceptionMessage(false) << std::endl;
|
||||
log << "An error occurred while processing the query " << "'" << query << "'"
|
||||
<< ": " << getCurrentExceptionMessage(false) << '\n';
|
||||
if (!(continue_on_errors || max_consecutive_errors > ++consecutive_errors))
|
||||
{
|
||||
shutdown = true;
|
||||
throw;
|
||||
}
|
||||
|
||||
std::cerr << getCurrentExceptionMessage(print_stacktrace,
|
||||
true /*check embedded stack trace*/) << std::endl;
|
||||
log << getCurrentExceptionMessage(print_stacktrace,
|
||||
true /*check embedded stack trace*/) << '\n' << flush;
|
||||
|
||||
size_t info_index = round_robin ? 0 : connection_index;
|
||||
++comparison_info_per_interval[info_index]->errors;
|
||||
@ -504,7 +503,7 @@ private:
|
||||
{
|
||||
std::lock_guard lock(mutex);
|
||||
|
||||
std::cerr << "\n";
|
||||
log << "\n";
|
||||
for (size_t i = 0; i < infos.size(); ++i)
|
||||
{
|
||||
const auto & info = infos[i];
|
||||
@ -524,31 +523,31 @@ private:
|
||||
connection_description += conn->getDescription();
|
||||
}
|
||||
}
|
||||
std::cerr
|
||||
<< connection_description << ", "
|
||||
<< "queries: " << info->queries << ", ";
|
||||
log
|
||||
<< connection_description << ", "
|
||||
<< "queries: " << info->queries.load() << ", ";
|
||||
if (info->errors)
|
||||
{
|
||||
std::cerr << "errors: " << info->errors << ", ";
|
||||
log << "errors: " << info->errors << ", ";
|
||||
}
|
||||
std::cerr
|
||||
<< "QPS: " << (info->queries / seconds) << ", "
|
||||
<< "RPS: " << (info->read_rows / seconds) << ", "
|
||||
<< "MiB/s: " << (info->read_bytes / seconds / 1048576) << ", "
|
||||
<< "result RPS: " << (info->result_rows / seconds) << ", "
|
||||
<< "result MiB/s: " << (info->result_bytes / seconds / 1048576) << "."
|
||||
<< "\n";
|
||||
log
|
||||
<< "QPS: " << fmt::format("{:.3f}", info->queries / seconds) << ", "
|
||||
<< "RPS: " << fmt::format("{:.3f}", info->read_rows / seconds) << ", "
|
||||
<< "MiB/s: " << fmt::format("{:.3f}", info->read_bytes / seconds / 1048576) << ", "
|
||||
<< "result RPS: " << fmt::format("{:.3f}", info->result_rows / seconds) << ", "
|
||||
<< "result MiB/s: " << fmt::format("{:.3f}", info->result_bytes / seconds / 1048576) << "."
|
||||
<< "\n";
|
||||
}
|
||||
std::cerr << "\n";
|
||||
log << "\n";
|
||||
|
||||
auto print_percentile = [&](double percent)
|
||||
{
|
||||
std::cerr << percent << "%\t\t";
|
||||
log << percent << "%\t\t";
|
||||
for (const auto & info : infos)
|
||||
{
|
||||
std::cerr << info->sampler.quantileNearest(percent / 100.0) << " sec.\t";
|
||||
log << fmt::format("{:.3f}", info->sampler.quantileNearest(percent / 100.0)) << " sec.\t";
|
||||
}
|
||||
std::cerr << "\n";
|
||||
log << "\n";
|
||||
};
|
||||
|
||||
for (int percent = 0; percent <= 90; percent += 10)
|
||||
@ -559,13 +558,15 @@ private:
|
||||
print_percentile(99.9);
|
||||
print_percentile(99.99);
|
||||
|
||||
std::cerr << "\n" << t_test.compareAndReport(confidence).second << "\n";
|
||||
log << "\n" << t_test.compareAndReport(confidence).second << "\n";
|
||||
|
||||
if (!cumulative)
|
||||
{
|
||||
for (auto & info : infos)
|
||||
info->clear();
|
||||
}
|
||||
|
||||
log.next();
|
||||
}
|
||||
|
||||
public:
|
||||
@ -741,7 +742,7 @@ int mainEntryClickHouseBenchmark(int argc, char ** argv)
|
||||
}
|
||||
catch (...)
|
||||
{
|
||||
std::cerr << getCurrentExceptionMessage(print_stacktrace, true) << std::endl;
|
||||
std::cerr << getCurrentExceptionMessage(print_stacktrace, true) << '\n';
|
||||
return getCurrentExceptionCode();
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1 @@
|
||||
0
|
||||
10
tests/queries/0_stateless/03271_benchmark_metrics.sh
Executable file
10
tests/queries/0_stateless/03271_benchmark_metrics.sh
Executable file
@ -0,0 +1,10 @@
|
||||
#!/usr/bin/env bash
|
||||
|
||||
set -e
|
||||
|
||||
CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
|
||||
# shellcheck source=../shell_config.sh
|
||||
. "$CUR_DIR"/../shell_config.sh
|
||||
|
||||
# A query with two seconds sleep cannot be processed with QPS > 0.5
|
||||
$CLICKHOUSE_BENCHMARK --query "SELECT sleep(2)" 2>&1 | grep -m1 -o -P 'QPS: \d+\.\d+' | $CLICKHOUSE_LOCAL --query "SELECT throwIf(extract(line, 'QPS: (.+)')::Float64 > 0.75) FROM table" --input-format LineAsString
|
||||
Loading…
Reference in New Issue
Block a user