ClickHouse/dbms/include/DB/DataStreams/IProfilingBlockInputStream.h

219 lines
11 KiB
C++
Raw Normal View History

2011-08-27 22:43:31 +00:00
#pragma once
#include <DB/Core/Progress.h>
2011-09-05 00:51:25 +00:00
#include <DB/Interpreters/Limits.h>
2015-01-18 08:25:56 +00:00
#include <DB/DataStreams/BlockStreamProfileInfo.h>
2011-08-27 22:43:31 +00:00
#include <DB/DataStreams/IBlockInputStream.h>
#include <atomic>
2011-08-27 22:43:31 +00:00
namespace DB
{
class QuotaForIntervals;
2016-03-07 04:35:06 +00:00
struct ProcessListElement;
class IProfilingBlockInputStream;
using ProfilingBlockInputStreamPtr = std::shared_ptr<IProfilingBlockInputStream>;
2011-08-27 22:43:31 +00:00
2012-05-09 08:16:09 +00:00
/** Смотрит за тем, как работает источник блоков.
2011-08-27 22:43:31 +00:00
* Позволяет получить информацию для профайлинга:
* строк в секунду, блоков в секунду, мегабайт в секунду и т. п.
2012-05-09 08:16:09 +00:00
* Позволяет остановить чтение данных (во вложенных источниках).
2011-08-27 22:43:31 +00:00
*/
2011-09-04 21:23:19 +00:00
class IProfilingBlockInputStream : public IBlockInputStream
2011-08-27 22:43:31 +00:00
{
public:
2015-01-18 08:25:56 +00:00
Block read() override final;
2011-09-04 21:23:19 +00:00
2016-05-31 01:04:34 +00:00
/** Реализация по-умолчанию вызывает readPrefixImpl() у себя, а затем readPrefix() у всех детей рекурсивно.
2016-05-20 20:43:07 +00:00
* Есть случаи, когда вы не хотите, чтобы readPrefix у детей вызывался синхронно, в этой функции,
* а хотите, чтобы они вызывались, например, в отдельных потоках (для распараллеливания инициализации детей).
* Тогда перегрузите функцию readPrefix.
*/
void readPrefix() override;
/** Реализация по-умолчанию вызывает рекурсивно readSuffix() у всех детей, а затем readSuffixImpl() у себя.
* Если этот поток вызывает у детей read() в отдельном потоке, этот поведение обычно неверно:
* readSuffix() у ребенка нельзя вызывать в момент, когда read() того же ребенка выполняется в другом потоке.
* В таком случае нужно переопределить этот метод, чтобы readSuffix() у детей вызывался, например, после соединения потоков.
*/
void readSuffix() override;
2013-09-13 20:33:09 +00:00
2016-08-13 02:21:17 +00:00
/// Get information about execution speed.
const BlockStreamProfileInfo & getProfileInfo() const { return info; }
2011-08-27 22:43:31 +00:00
/** Получить "тотальные" значения.
* Реализация по-умолчанию берёт их из себя или из первого дочернего источника, в котором они есть.
* Переопределённый метод может провести некоторые вычисления. Например, применить выражение к totals дочернего источника.
* Тотальных значений может не быть - тогда возвращается пустой блок.
*
* Вызывайте этот метод только после получения всех данных с помощью read,
* иначе будут проблемы, если какие-то данные в это же время вычисляются в другом потоке.
*/
virtual const Block & getTotals();
2014-09-12 16:05:29 +00:00
/// То же самое для минимумов и максимумов.
const Block & getExtremes() const;
2012-10-20 05:54:35 +00:00
2012-10-18 19:24:46 +00:00
/** Установить колбэк прогресса выполнения.
* Колбэк пробрасывается во все дочерние источники.
2012-10-18 19:24:46 +00:00
* По-умолчанию, он вызывается для листовых источников, после каждого блока.
* (Но это может быть переопределено в методе progress())
* Функция принимает количество строк в последнем блоке, количество байт в последнем блоке.
* Следует иметь ввиду, что колбэк может вызываться из разных потоков.
*/
void setProgressCallback(ProgressCallback callback);
/** В этом методе:
* - вызывается колбэк прогресса;
* - обновляется статус выполнения запроса в ProcessList-е;
* - проверяются ограничения и квоты, которые должны быть проверены не в рамках одного источника,
* а над общим количеством потраченных ресурсов во всех источниках сразу (информация в ProcessList-е).
*/
virtual void progress(const Progress & value)
{
/// Данные для прогресса берутся из листовых источников.
if (children.empty())
progressImpl(value);
}
void progressImpl(const Progress & value);
/** Установить указатель на элемент списка процессов.
* Пробрасывается во все дочерние источники.
* В него будет записываться общая информация о потраченных на запрос ресурсах.
* На основе этой информации будет проверяться квота, и некоторые ограничения.
* Также эта информация будет доступна в запросе SHOW PROCESSLIST.
*/
void setProcessListElement(ProcessListElement * elem);
2012-10-18 19:24:46 +00:00
/** Установить информацию о приблизительном общем количестве строк, которых нужно прочитать.
*/
void setTotalRowsApprox(size_t value) { total_rows_approx = value; }
2012-10-18 19:24:46 +00:00
/** Попросить прервать получение данных как можно скорее.
* По-умолчанию - просто выставляет флаг is_cancelled и просит прерваться всех детей.
2012-11-10 05:13:46 +00:00
* Эта функция может вызываться несколько раз, в том числе, одновременно из разных потоков.
*/
virtual void cancel();
2012-10-18 19:24:46 +00:00
/** Требуется ли прервать получение данных.
2012-10-20 05:54:35 +00:00
*/
bool isCancelled() const
2012-10-18 19:24:46 +00:00
{
return is_cancelled.load(std::memory_order_seq_cst);
2012-10-18 19:24:46 +00:00
}
2012-05-17 19:15:53 +00:00
/** Какие ограничения (и квоты) проверяются.
* Если LIMITS_CURRENT - ограничения проверяются на количество данных, прочитанных только в этом stream-е.
* - используется для реализации ограничений на объём результата выполнения запроса.
* Если LIMITS_TOTAL, то ещё дополнительно делается проверка в колбэке прогресса,
* по суммарным данным по всем листовым stream-ам, в том числе, с удалённых серверов.
* - используется для реализации ограничений на общий объём прочитанных (исходных) данных.
*/
enum LimitsMode
{
LIMITS_CURRENT,
LIMITS_TOTAL,
};
/// Используется подмножество ограничений из Limits.
struct LocalLimits
{
LimitsMode mode = LIMITS_CURRENT;
2014-09-12 16:05:29 +00:00
size_t max_rows_to_read = 0;
size_t max_bytes_to_read = 0;
OverflowMode read_overflow_mode = OverflowMode::THROW;
Poco::Timespan max_execution_time = 0;
OverflowMode timeout_overflow_mode = OverflowMode::THROW;
/// В строчках в секунду.
size_t min_execution_speed = 0;
/// Проверять, что скорость не слишком низкая, после прошествия указанного времени.
Poco::Timespan timeout_before_checking_execution_speed = 0;
};
/** Установить ограничения для проверки на каждый блок. */
void setLimits(const LocalLimits & limits_)
{
limits = limits_;
}
/** Установить квоту. Если устанавливается квота на объём исходных данных,
* то следует ещё установить mode = LIMITS_TOTAL в LocalLimits с помощью setLimits.
*/
void setQuota(QuotaForIntervals & quota_)
2013-08-28 20:47:22 +00:00
{
quota = &quota_;
}
/// Включить рассчёт минимумов и максимумов по столбцам результата.
void enableExtremes() { enabled_extremes = true; }
2012-05-21 20:38:34 +00:00
protected:
2011-08-27 22:43:31 +00:00
BlockStreamProfileInfo info;
std::atomic<bool> is_cancelled{false};
2012-05-09 15:15:45 +00:00
ProgressCallback progress_callback;
ProcessListElement * process_list_elem = nullptr;
2012-10-20 02:10:47 +00:00
2014-04-08 07:31:51 +00:00
bool enabled_extremes = false;
/// Дополнительная информация, которая может образоваться в процессе работы.
/// Тотальные значения при агрегации.
Block totals;
/// Минимумы и максимумы. Первая строчка блока - минимумы, вторая - максимумы.
Block extremes;
/// Приблизительное общее количество строк, которых нужно прочитать. Для прогресс-бара.
size_t total_rows_approx = 0;
/// Информация о приблизительном общем количестве строк собрана в родительском источнике.
bool collected_total_rows_approx = false;
/// Превышено ограничение на количество строк/байт, и нужно прекратить выполнение на следующем вызове read, как будто поток иссяк.
bool limit_exceeded_need_break = false;
/// Ограничения и квоты.
2014-09-12 16:05:29 +00:00
LocalLimits limits;
2014-04-08 07:31:51 +00:00
QuotaForIntervals * quota = nullptr; /// Если nullptr - квота не используется.
double prev_elapsed = 0;
2013-08-28 20:47:22 +00:00
2012-10-20 02:10:47 +00:00
/// Наследники должны реализовать эту функцию.
virtual Block readImpl() = 0;
/// Здесь можно делать предварительную инициализацию.
virtual void readPrefixImpl() {}
2013-09-13 20:33:09 +00:00
/// Здесь необходимо делать финализацию, которая может привести к исключению.
virtual void readSuffixImpl() {}
void updateExtremes(Block & block);
/** Проверить ограничения и квоты.
* Но только те, что могут быть проверены в рамках каждого отдельного источника.
*/
bool checkLimits();
void checkQuota(Block & block);
/// Собрать информацию о приблизительном общем числе строк по всем детям.
void collectTotalRowsApprox();
/** Передать информацию о приблизительном общем числе строк в колбэк прогресса.
* Сделано так, что отправка происходит лишь в верхнем источнике.
*/
void collectAndSendTotalRowsApprox();
2011-08-27 22:43:31 +00:00
};
}