1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
#include <QueryPipeline/RemoteInserter.h>
#include <Client/Connection.h>
#include <Common/logger_useful.h>
#include <Common/NetException.h>
#include <Common/CurrentThread.h>
#include <Interpreters/InternalTextLogsQueue.h>
#include <IO/ConnectionTimeouts.h>
#include <Core/Settings.h>
namespace DB
{
namespace ErrorCodes
{
extern const int UNEXPECTED_PACKET_FROM_SERVER;
}
RemoteInserter::RemoteInserter(
Connection & connection_,
const ConnectionTimeouts & timeouts,
const String & query_,
const Settings & settings_,
const ClientInfo & client_info_)
: connection(connection_)
, query(query_)
, server_revision(connection.getServerRevision(timeouts))
{
ClientInfo modified_client_info = client_info_;
modified_client_info.query_kind = ClientInfo::QueryKind::SECONDARY_QUERY;
Settings settings = settings_;
/// With current protocol it is impossible to avoid deadlock in case of send_logs_level!=none.
///
/// RemoteInserter send Data blocks/packets to the remote shard,
/// while remote side can send Log packets to the initiator (this RemoteInserter instance).
///
/// But it is not enough to pull Log packets just before writing the next block
/// since there is no way to ensure that all Log packets had been consumed.
///
/// And if enough Log packets will be queued by the remote side,
/// it will wait send_timeout until initiator will consume those packets,
/// while initiator already starts writing Data blocks,
/// and will not consume Log packets.
///
/// So that is why send_logs_level had been disabled here.
settings.send_logs_level = "none";
/** Send query and receive "header", that describes table structure.
* Header is needed to know, what structure is required for blocks to be passed to 'write' method.
*/
connection.sendQuery(
timeouts, query, /* query_parameters */ {}, "", QueryProcessingStage::Complete, &settings, &modified_client_info, false, {});
while (true)
{
Packet packet = connection.receivePacket();
if (Protocol::Server::Data == packet.type)
{
header = packet.block;
break;
}
else if (Protocol::Server::Exception == packet.type)
{
packet.exception->rethrow();
break;
}
else if (Protocol::Server::Log == packet.type)
{
/// Pass logs from remote server to client
if (auto log_queue = CurrentThread::getInternalTextLogsQueue())
log_queue->pushBlock(std::move(packet.block));
}
else if (Protocol::Server::TableColumns == packet.type)
{
/// Server could attach ColumnsDescription in front of stream for column defaults. There's no need to pass it through cause
/// client's already got this information for remote table. Ignore.
}
else
throw NetException(
ErrorCodes::UNEXPECTED_PACKET_FROM_SERVER,
"Unexpected packet from server (expected Data or Exception, got {})",
Protocol::Server::toString(packet.type));
}
}
void RemoteInserter::write(Block block)
{
try
{
connection.sendData(block, /* name */"", /* scalar */false);
}
catch (const NetException &)
{
/// Try to get more detailed exception from server
auto packet_type = connection.checkPacket(/* timeout_microseconds */0);
if (packet_type && *packet_type == Protocol::Server::Exception)
{
Packet packet = connection.receivePacket();
packet.exception->rethrow();
}
throw;
}
}
void RemoteInserter::writePrepared(ReadBuffer & buf, size_t size)
{
/// We cannot use 'header'. Input must contain block with proper structure.
connection.sendPreparedData(buf, size);
}
void RemoteInserter::onFinish()
{
/// Empty block means end of data.
connection.sendData(Block(), /* name */"", /* scalar */false);
/// Wait for EndOfStream or Exception packet, skip Log packets.
while (true)
{
Packet packet = connection.receivePacket();
if (Protocol::Server::EndOfStream == packet.type)
break;
else if (Protocol::Server::Exception == packet.type)
packet.exception->rethrow();
else if (Protocol::Server::Log == packet.type || Protocol::Server::TimezoneUpdate == packet.type)
{
// Do nothing
}
else
throw NetException(
ErrorCodes::UNEXPECTED_PACKET_FROM_SERVER,
"Unexpected packet from server (expected EndOfStream or Exception, got {})",
Protocol::Server::toString(packet.type));
}
finished = true;
}
RemoteInserter::~RemoteInserter()
{
/// If interrupted in the middle of the loop of communication with the server, then interrupt the connection,
/// to not leave the connection in unsynchronized state.
if (!finished)
{
try
{
connection.disconnect();
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
}
}
|
