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

172 lines
7.5 KiB
C++
Raw Normal View History

2012-05-17 19:15:53 +00:00
#pragma once
2015-09-29 19:19:54 +00:00
#include <common/logger_useful.h>
2012-05-17 19:15:53 +00:00
#include <DB/DataStreams/IProfilingBlockInputStream.h>
#include <DB/Common/Throttler.h>
#include <DB/Interpreters/Context.h>
#include <DB/Client/ConnectionPool.h>
2015-11-06 17:44:01 +00:00
#include <DB/Client/MultiplexedConnections.h>
#include <DB/Interpreters/Cluster.h>
2012-05-17 19:15:53 +00:00
namespace DB
{
2015-11-06 17:44:01 +00:00
/** Позволяет выполнить запрос на удалённых репликах одного шарда и получить результат.
2012-05-17 19:15:53 +00:00
*/
class RemoteBlockInputStream : public IProfilingBlockInputStream
{
public:
/// Принимает готовое соединение.
2015-11-06 17:44:01 +00:00
RemoteBlockInputStream(Connection & connection_, const String & query_, const Settings * settings_,
ThrottlerPtr throttler_ = nullptr, const Tables & external_tables_ = Tables(),
QueryProcessingStage::Enum stage_ = QueryProcessingStage::Complete,
const Context & context_ = getDefaultContext());
/// Принимает готовое соединение. Захватывает владение соединением из пула.
2015-11-06 17:44:01 +00:00
RemoteBlockInputStream(ConnectionPool::Entry & pool_entry_, const String & query_, const Settings * settings_,
ThrottlerPtr throttler_ = nullptr, const Tables & external_tables_ = Tables(),
QueryProcessingStage::Enum stage_ = QueryProcessingStage::Complete,
const Context & context_ = getDefaultContext());
/// Принимает пул, из которого нужно будет достать одно или несколько соединений.
RemoteBlockInputStream(ConnectionPoolPtr & pool_, const String & query_, const Settings * settings_,
2015-11-06 17:44:01 +00:00
ThrottlerPtr throttler_ = nullptr, const Tables & external_tables_ = Tables(),
QueryProcessingStage::Enum stage_ = QueryProcessingStage::Complete,
const Context & context_ = getDefaultContext());
2015-11-11 02:13:24 +00:00
/// Принимает пулы - один для каждого шарда, из которых нужно будет достать одно или несколько соединений.
2015-11-06 17:44:01 +00:00
RemoteBlockInputStream(ConnectionPoolsPtr & pools_, const String & query_, const Settings * settings_,
ThrottlerPtr throttler_ = nullptr, const Tables & external_tables_ = Tables(),
QueryProcessingStage::Enum stage_ = QueryProcessingStage::Complete,
const Context & context_ = getDefaultContext());
2012-10-18 19:24:46 +00:00
2015-11-06 17:44:01 +00:00
~RemoteBlockInputStream() override;
2012-05-17 19:15:53 +00:00
/// Specify how we allocate connections on a shard.
2016-03-01 17:47:53 +00:00
void setPoolMode(PoolMode pool_mode_);
2012-05-17 19:15:53 +00:00
2015-11-06 17:44:01 +00:00
/// Кроме блоков, получить информацию о блоках.
void appendExtraInfo();
2015-11-06 17:44:01 +00:00
/// Отправляет запрос (инициирует вычисления) раньше, чем read.
void readPrefix() override;
2012-10-18 19:24:46 +00:00
2012-05-17 19:15:53 +00:00
/** Отменяем умолчальное уведомление о прогрессе,
* так как колбэк прогресса вызывается самостоятельно.
*/
void progress(const Progress & value) override {}
2012-05-17 19:15:53 +00:00
2015-11-06 17:44:01 +00:00
void cancel() override;
2012-10-18 19:24:46 +00:00
2015-11-06 17:44:01 +00:00
String getName() const override { return "Remote"; }
2015-10-12 14:53:16 +00:00
2015-11-06 17:44:01 +00:00
String getID() const override
2015-10-12 14:53:16 +00:00
{
2015-11-06 17:44:01 +00:00
std::stringstream res;
res << this;
return res.str();
2015-10-12 14:53:16 +00:00
}
BlockExtraInfo getBlockExtraInfo() const override
{
2015-11-06 17:44:01 +00:00
return multiplexed_connections->getBlockExtraInfo();
2015-10-12 14:53:16 +00:00
}
2012-10-20 02:10:47 +00:00
protected:
/// Отправить на удаленные серверы все временные таблицы.
2015-11-06 17:44:01 +00:00
void sendExternalTables();
2015-11-06 17:44:01 +00:00
Block readImpl() override;
2015-11-06 17:44:01 +00:00
void readSuffixImpl() override;
2013-09-13 20:33:09 +00:00
/// Создать объект для общения с репликами одного шарда, на которых должен выполниться запрос.
2015-11-06 17:44:01 +00:00
void createMultiplexedConnections();
/// Возвращает true, если запрос отправлен.
2015-11-06 17:44:01 +00:00
bool isQueryPending() const;
/// Возвращает true, если исключение было выкинуто.
2015-11-06 17:44:01 +00:00
bool hasThrownException() const;
private:
2015-11-06 17:44:01 +00:00
void init(const Settings * settings_);
2015-11-06 17:44:01 +00:00
void sendQuery();
2015-11-06 17:44:01 +00:00
/// Отправить запрос на отмену всех соединений к репликам, если такой запрос ещё не был отправлен.
void tryCancel(const char * reason);
/// ITable::read requires a Context, therefore we should create one if the user can't supply it
static Context & getDefaultContext()
{
static Context instance;
return instance;
}
2012-05-17 19:15:53 +00:00
private:
2015-11-06 17:44:01 +00:00
/// Готовое соединение.
ConnectionPool::Entry pool_entry;
Connection * connection = nullptr;
2015-11-06 17:44:01 +00:00
/// Пул соединений одного шарда.
ConnectionPoolPtr pool = nullptr;
2015-11-06 17:44:01 +00:00
/// Пулы соединений одного или нескольких шардов.
ConnectionPoolsPtr pools;
std::unique_ptr<MultiplexedConnections> multiplexed_connections;
2012-05-23 19:51:30 +00:00
const String query;
bool send_settings;
Settings settings;
/// Если не nullptr, то используется, чтобы ограничить сетевой трафик.
ThrottlerPtr throttler;
/// Временные таблицы, которые необходимо переслать на удаленные сервера.
Tables external_tables;
2012-05-17 19:15:53 +00:00
QueryProcessingStage::Enum stage;
Context context;
2012-05-17 19:15:53 +00:00
/// Потоки для чтения из временных таблиц - для последующей отправки данных на удалённые серверы для GLOBAL-подзапросов.
std::vector<ExternalTablesData> external_tables_data;
std::mutex external_tables_mutex;
/// Установили соединения с репликами, но ещё не отправили запрос.
std::atomic<bool> established { false };
2012-10-18 19:00:49 +00:00
/// Отправили запрос (это делается перед получением первого блока).
std::atomic<bool> sent_query { false };
/** Получили все данные от всех реплик, до пакета EndOfStream.
2012-10-18 19:00:49 +00:00
* Если при уничтожении объекта, ещё не все данные считаны,
* то для того, чтобы не было рассинхронизации, на реплики отправляются просьбы прервать выполнение запроса,
2012-10-18 19:00:49 +00:00
* и после этого считываются все пакеты до EndOfStream.
*/
std::atomic<bool> finished { false };
/** На каждую реплику была отправлена просьба прервать выполнение запроса, так как данные больше не нужны.
2012-10-18 19:00:49 +00:00
* Это может быть из-за того, что данных достаточно (например, при использовании LIMIT),
* или если на стороне клиента произошло исключение.
*/
std::atomic<bool> was_cancelled { false };
/** С одной репилки было получено исключение. В этом случае получать больше пакетов или
* просить прервать запрос на этой реплике не нужно.
*/
std::atomic<bool> got_exception_from_replica { false };
/** С одной реплики был получен неизвестный пакет. В этом случае получать больше пакетов или
* просить прервать запрос на этой реплике не нужно.
*/
std::atomic<bool> got_unknown_packet_from_replica { false };
2012-10-18 19:00:49 +00:00
2015-10-12 14:53:16 +00:00
bool append_extra_info = false;
2016-03-01 17:47:53 +00:00
PoolMode pool_mode = PoolMode::GET_MANY;
2015-10-12 14:53:16 +00:00
Logger * log = &Logger::get("RemoteBlockInputStream");
2012-05-17 19:15:53 +00:00
};
}