Skip to content

Commit b7be415

Browse files
alexey-milovidovalexey-milovidov
authored
Merge pull request #419 from pet74alex/support-external-data
Support for external data
2 parents fbd7945 + be7f402 commit b7be415

File tree

3 files changed

+120
-12
lines changed

3 files changed

+120
-12
lines changed

‎README.md‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,33 @@ int main()
103103
}
104104
);
105105

106+
/// Select values inserted in the previous step using external data feature
107+
/// See https://clickhouse.com/docs/engines/table-engines/special/external-data
108+
{
109+
Block block1, block2;
110+
auto id = std::make_shared<ColumnUInt64>();
111+
id->Append(1);
112+
block1.AppendColumn("id" , id);
113+
114+
auto name = std::make_shared<ColumnString>();
115+
name->Append("seven");
116+
block2.AppendColumn("name", name);
117+
118+
const std::string _1 = "_1";
119+
const std::string _2 = "_2";
120+
121+
const ExternalTables external = {{_1, block1}, {_2, block2}};
122+
client.SelectWithExternalData("SELECT id, name FROM default.numbers where id in (_1) or name in (_2)",
123+
external, [] (const Block& block)
124+
{
125+
for (size_t i = 0; i < block.GetRowCount(); ++i) {
126+
std::cout << block[0]->As<ColumnUInt64>()->At(i) << " "
127+
<< block[1]->As<ColumnString>()->At(i) << "\n";
128+
}
129+
}
130+
);
131+
}
132+
106133
/// Delete table.
107134
client.Execute("DROP TABLE default.numbers");
108135

‎clickhouse/client.cpp‎

Lines changed: 76 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,8 @@ class Client::Impl {
155155

156156
void ExecuteQuery(Query query);
157157

158+
void SelectWithExternalData(Query query, const ExternalTables& external_tables);
159+
158160
void SendCancel();
159161

160162
void Insert(const std::string& table_name, const std::string& query_id, const Block& block);
@@ -174,10 +176,14 @@ class Client::Impl {
174176

175177
bool ReceivePacket(uint64_t* server_packet = nullptr);
176178

177-
void SendQuery(const Query& query);
179+
void SendQuery(const Query& query, bool finalize = true);
180+
void FinalizeQuery();
178181

179182
void SendData(const Block& block);
180183

184+
void SendBlockData(const Block& block);
185+
void SendExternalData(const ExternalTables& external_tables);
186+
181187
bool SendHello();
182188

183189
bool ReadBlock(InputStream& input, Block* block);
@@ -291,6 +297,51 @@ void Client::Impl::ExecuteQuery(Query query) {
291297
}
292298
}
293299

300+
301+
void Client::Impl::SelectWithExternalData(Query query, const ExternalTables& external_tables) {
302+
if (server_info_.revision < DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES) {
303+
throw UnimplementedError("This version of ClickHouse server doesn't support temporary tables");
304+
}
305+
306+
EnsureNull en(static_cast<QueryEvents*>(&query), &events_);
307+
308+
if (options_.ping_before_query) {
309+
RetryGuard([this]() { Ping(); });
310+
}
311+
312+
SendQuery(query, false);
313+
SendExternalData(external_tables);
314+
FinalizeQuery();
315+
316+
while (ReceivePacket()) {
317+
;
318+
}
319+
}
320+
321+
void Client::Impl::SendBlockData(const Block& block) {
322+
if (compression_ == CompressionState::Enable) {
323+
std::unique_ptr<OutputStream> compressed_output = std::make_unique<CompressedOutput>(output_.get(), options_.max_compression_chunk_size, options_.compression_method);
324+
BufferedOutput buffered(std::move(compressed_output), options_.max_compression_chunk_size);
325+
326+
WriteBlock(block, buffered);
327+
} else {
328+
WriteBlock(block, *output_);
329+
}
330+
}
331+
332+
void Client::Impl::SendExternalData(const ExternalTables& external_tables) {
333+
for (const auto& table: external_tables) {
334+
if (!table.data.GetRowCount()) {
335+
// skip empty blocks to keep the connection in the consistent state as the current request would be marked as finished by such an empty block
336+
continue;
337+
}
338+
WireFormat::WriteFixed<uint8_t>(*output_, ClientCodes::Data);
339+
WireFormat::WriteString(*output_, table.name);
340+
SendBlockData(table.data);
341+
}
342+
}
343+
344+
294345
std::string NameToQueryString(const std::string &input)
295346
{
296347
std::string output;
@@ -753,7 +804,7 @@ void Client::Impl::SendCancel() {
753804
output_->Flush();
754805
}
755806

756-
void Client::Impl::SendQuery(const Query& query) {
807+
void Client::Impl::SendQuery(const Query& query, bool finalize) {
757808
WireFormat::WriteUInt64(*output_, ClientCodes::Query);
758809
WireFormat::WriteString(*output_, query.GetQueryID());
759810

@@ -858,7 +909,13 @@ void Client::Impl::SendQuery(const Query& query) {
858909
}
859910
WireFormat::WriteString(*output_, std::string()); // empty string after last param
860911
}
912+
913+
if (finalize) {
914+
FinalizeQuery();
915+
}
916+
}
861917

918+
void Client::Impl::FinalizeQuery() {
862919
// Send empty block as marker of
863920
// end of data
864921
SendData(Block());
@@ -905,16 +962,7 @@ void Client::Impl::SendData(const Block& block) {
905962
if (server_info_.revision >= DBMS_MIN_REVISION_WITH_TEMPORARY_TABLES) {
906963
WireFormat::WriteString(*output_, std::string());
907964
}
908-
909-
if (compression_ == CompressionState::Enable) {
910-
911-
std::unique_ptr<OutputStream> compressed_output = std::make_unique<CompressedOutput>(output_.get(), options_.max_compression_chunk_size, options_.compression_method);
912-
BufferedOutput buffered(std::move(compressed_output), options_.max_compression_chunk_size);
913-
914-
WriteBlock(block, buffered);
915-
} else {
916-
WriteBlock(block, *output_);
917-
}
965+
SendBlockData(block);
918966

919967
output_->Flush();
920968
}
@@ -1077,6 +1125,22 @@ void Client::Select(const Query& query) {
10771125
Execute(query);
10781126
}
10791127

1128+
void Client::SelectWithExternalData(const std::string& query, const ExternalTables& external_tables, SelectCallback cb) {
1129+
impl_->SelectWithExternalData(Query(query).OnData(std::move(cb)), external_tables);
1130+
}
1131+
1132+
void Client::SelectWithExternalData(const std::string& query, const std::string& query_id, const ExternalTables& external_tables, SelectCallback cb) {
1133+
impl_->SelectWithExternalData(Query(query, query_id).OnData(std::move(cb)), external_tables);
1134+
}
1135+
1136+
void Client::SelectWithExternalDataCancelable(const std::string& query, const ExternalTables& external_tables, SelectCancelableCallback cb) {
1137+
impl_->SelectWithExternalData(Query(query).OnDataCancelable(std::move(cb)), external_tables);
1138+
}
1139+
1140+
void Client::SelectWithExternalDataCancelable(const std::string& query, const std::string& query_id, const ExternalTables& external_tables, SelectCancelableCallback cb) {
1141+
impl_->SelectWithExternalData(Query(query, query_id).OnDataCancelable(std::move(cb)), external_tables);
1142+
}
1143+
10801144
void Client::Insert(const std::string& table_name, const Block& block) {
10811145
impl_->Insert(table_name, Query::default_query_id, block);
10821146
}

‎clickhouse/client.h‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -225,6 +225,13 @@ std::ostream& operator<<(std::ostream& os, const Endpoint& options);
225225

226226
class SocketFactory;
227227

228+
struct ExternalTable {
229+
const std::string_view name;
230+
const Block& data;
231+
};
232+
233+
using ExternalTables = std::vector<ExternalTable>;
234+
228235
/**
229236
*
230237
*/
@@ -248,6 +255,16 @@ class Client {
248255
void SelectCancelable(const std::string& query, SelectCancelableCallback cb);
249256
void SelectCancelable(const std::string& query, const std::string& query_id, SelectCancelableCallback cb);
250257

258+
// The same as Select but with an external data
259+
// required for the query, see https://clickhouse.com/docs/engines/table-engines/special/external-data
260+
void SelectWithExternalData(const std::string& query, const ExternalTables& external_tables, SelectCallback cb);
261+
void SelectWithExternalData(const std::string& query, const std::string& query_id, const ExternalTables& external_tables, SelectCallback cb);
262+
263+
// The same as SelectWithExternalData but can be canceled by returning false from
264+
// the data handler function \p cb.
265+
void SelectWithExternalDataCancelable(const std::string& query, const ExternalTables& external_tables, SelectCancelableCallback cb);
266+
void SelectWithExternalDataCancelable(const std::string& query, const std::string& query_id, const ExternalTables& external_tables, SelectCancelableCallback cb);
267+
251268
/// Alias for Execute.
252269
void Select(const Query& query);
253270

0 commit comments

Comments
 (0)