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

85 lines
2.2 KiB
C
Raw Normal View History

2012-05-21 20:38:34 +00:00
#pragma once
#include <DB/DataStreams/IBlockOutputStream.h>
#include <DB/Client/Connection.h>
namespace DB
{
/** Позволяет выполнить запрос INSERT на удалённом сервере и отправить данные.
*/
class RemoteBlockOutputStream : public IBlockOutputStream
{
public:
RemoteBlockOutputStream(Connection & connection_, const String & query_)
: connection(connection_), query(query_)
{
}
/** Отправляет запрос и получает блок-пример, описывающий структуру таблицы.
* Он нужен, чтобы знать, какие блоки передавать в метод write.
* Вызывайте только перед write.
*/
Block sendQueryAndGetSampleBlock()
{
2012-06-07 20:02:41 +00:00
connection.sendQuery(query);
2012-05-21 20:38:34 +00:00
sent_query = true;
Connection::Packet packet = connection.receivePacket();
if (Protocol::Server::Data == packet.type)
return packet.block;
2012-06-19 22:46:02 +00:00
else if (Protocol::Server::Exception == packet.type)
{
packet.exception->rethrow();
return Block();
}
2012-05-21 20:38:34 +00:00
else
2012-06-19 22:46:02 +00:00
throw Exception("Unexpected packet from server (expected Data or Exception, got "
+ String(Protocol::Server::toString(Protocol::Server::Enum(packet.type))) + ")", ErrorCodes::UNEXPECTED_PACKET_FROM_SERVER);
2012-05-21 20:38:34 +00:00
}
void write(const Block & block)
{
if (!sent_query)
sendQueryAndGetSampleBlock();
connection.sendData(block);
}
void writeSuffix()
{
/// Пустой блок означает конец данных.
connection.sendData(Block());
/// Получаем пакет EndOfStream.
Connection::Packet packet = connection.receivePacket();
2012-06-19 22:46:02 +00:00
if (Protocol::Server::EndOfStream == packet.type)
{
/// Ничего.
}
else if (Protocol::Server::Exception == packet.type)
packet.exception->rethrow();
else
throw Exception("Unexpected packet from server (expected EndOfStream or Exception, got "
+ String(Protocol::Server::toString(Protocol::Server::Enum(packet.type))) + ")", ErrorCodes::UNEXPECTED_PACKET_FROM_SERVER);
2012-05-21 20:38:34 +00:00
}
BlockOutputStreamPtr clone() { return new RemoteBlockOutputStream(connection, query); }
private:
Connection & connection;
2012-06-07 20:02:41 +00:00
String query;
2012-05-21 20:38:34 +00:00
bool sent_query;
};
}