{

try

{

if (connected)

disconnect();

LOG_TRACE(log_wrapper.get(), "Connecting. Database: {}. User: {}{}{}",

default_database.empty() ? "(not specified)" : default_database,

user,

static_cast<bool>(secure) ? ". Secure" : "",

static_cast<bool>(compression) ? "" : ". Uncompressed");

if (static_cast<bool>(secure))

{

#if USE_SSL

socket = std::make_unique<Poco::Net::SecureStreamSocket>();

/// we resolve the ip when we open SecureStreamSocket, so to make Server Name Indication (SNI)

/// work we need to pass host name separately. It will be send into TLS Hello packet to let

/// the server know which host we want to talk with (single IP can process requests for multiple hosts using SNI).

static_cast<Poco::Net::SecureStreamSocket*>(socket.get())->setPeerHostName(host);

#else

throw Exception{"tcp_secure protocol is disabled because poco library was built without NetSSL support.", ErrorCodes::SUPPORT_IS_DISABLED};

#endif

}

else

{

socket = std::make_unique<Poco::Net::StreamSocket>();

}

current_resolved_address = DNSResolver::instance().resolveAddress(host, port);

const auto & connection_timeout = static_cast<bool>(secure) ? timeouts.secure_connection_timeout : timeouts.connection_timeout;

socket->connect(*current_resolved_address, connection_timeout);

socket->setReceiveTimeout(timeouts.receive_timeout);

socket->setSendTimeout(timeouts.send_timeout);

socket->setNoDelay(true);

if (timeouts.tcp_keep_alive_timeout.totalSeconds())

{

socket->setKeepAlive(true);

socket->setOption(IPPROTO_TCP,

#if defined(TCP_KEEPALIVE)

TCP_KEEPALIVE

#else

TCP_KEEPIDLE // __APPLE__

#endif

, timeouts.tcp_keep_alive_timeout);

}

in = std::make_shared<ReadBufferFromPocoSocket>(*socket);

out = std::make_shared<WriteBufferFromPocoSocket>(*socket);

connected = true;

sendHello();

receiveHello();

LOG_TRACE(log_wrapper.get(), "Connected to {} server version {}.{}.{}.",

server_name, server_version_major, server_version_minor, server_version_patch);

}

catch (Poco::Net::NetException & e)

{

disconnect();

/// Add server address to exception. Also Exception will remember stack trace. It's a pity that more precise exception type is lost.

throw NetException(e.displayText() + " (" + getDescription() + ")", ErrorCodes::NETWORK_ERROR);

}

catch (Poco::TimeoutException & e)

{

disconnect();

/// Add server address to exception. Also Exception will remember stack trace. It's a pity that more precise exception type is lost.

throw NetException(e.displayText() + " (" + getDescription() + ")", ErrorCodes::SOCKET_TIMEOUT);

}

{

in = nullptr;

last_input_packet_type.reset();

out = nullptr; // can write to socket

if (socket)

socket->close();

socket = nullptr;

connected = false;

{

/** Disallow control characters in user controlled parameters

auto has_control_character = [](const std::string & s)

{

for (auto c : s)

if (isControlASCII(c))

return true;

return false;

};

if (has_control_character(default_database)

|| has_control_character(user)

|| has_control_character(password))

throw Exception("Parameters 'default_database', 'user' and 'password' must not contain ASCII control characters", ErrorCodes::BAD_ARGUMENTS);

writeVarUInt(Protocol::Client::Hello, *out);

writeStringBinary((DBMS_NAME " ") + client_name, *out);

writeVarUInt(DBMS_VERSION_MAJOR, *out);

writeVarUInt(DBMS_VERSION_MINOR, *out);

// NOTE For backward compatibility of the protocol, client cannot send its version_patch.

writeVarUInt(DBMS_TCP_PROTOCOL_VERSION, *out);

writeStringBinary(default_database, *out);

/// If interserver-secret is used, one do not need password

/// (NOTE we do not check for DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET, since we cannot ignore inter-server secret if it was requested)

if (!cluster_secret.empty())

{

writeStringBinary(USER_INTERSERVER_MARKER, *out);

writeStringBinary("" /* password */, *out);

#if USE_SSL

sendClusterNameAndSalt();

#else

throw Exception(

"Inter-server secret support is disabled, because ClickHouse was built without SSL library",

ErrorCodes::SUPPORT_IS_DISABLED);

#endif

}

else

{

writeStringBinary(user, *out);

writeStringBinary(password, *out);

}

out->next();

{

/// Receive hello packet.

UInt64 packet_type = 0;

readVarUInt(packet_type, *in);

if (packet_type == Protocol::Server::Hello)

{

readStringBinary(server_name, *in);

readVarUInt(server_version_major, *in);

readVarUInt(server_version_minor, *in);

readVarUInt(server_revision, *in);

if (server_revision >= DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE)

readStringBinary(server_timezone, *in);

if (server_revision >= DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME)

readStringBinary(server_display_name, *in);

if (server_revision >= DBMS_MIN_REVISION_WITH_VERSION_PATCH)

readVarUInt(server_version_patch, *in);

else

server_version_patch = server_revision;

}

else if (packet_type == Protocol::Server::Exception)

receiveException()->rethrow();

else

{

/// Close connection, to not stay in unsynchronised state.

disconnect();

throwUnexpectedPacket(packet_type, "Hello or Exception");

}

String & name,

UInt64 & version_major,

UInt64 & version_minor,

UInt64 & version_patch,

UInt64 & revision)

{

if (!connected)

connect(timeouts);

name = server_name;

version_major = server_version_major;

version_minor = server_version_minor;

version_patch = server_version_patch;

revision = server_revision;

{

if (!connected)

connect(timeouts);

return server_revision;

{

if (!connected)

{

connect(timeouts);

}

else if (!ping())

{

LOG_TRACE(log_wrapper.get(), "Connection was closed, will reconnect.");

connect(timeouts);

}

{

pcg64_fast rng(randomSeed());

UInt64 rand = rng();

salt = encodeSHA256(&rand, sizeof(rand));

writeStringBinary(cluster, *out);

writeStringBinary(salt, *out);

{

// LOG_TRACE(log_wrapper.get(), "Ping");

TimeoutSetter timeout_setter(*socket, sync_request_timeout, true);

try

{

UInt64 pong = 0;

writeVarUInt(Protocol::Client::Ping, *out);

out->next();

if (in->eof())

return false;

readVarUInt(pong, *in);

/// Could receive late packets with progress. TODO: Maybe possible to fix.

while (pong == Protocol::Server::Progress)

{

receiveProgress();

if (in->eof())

return false;

readVarUInt(pong, *in);

}

if (pong != Protocol::Server::Pong)

throwUnexpectedPacket(pong, "Pong");

}

catch (const Poco::Exception & e)

{

LOG_TRACE(log_wrapper.get(), e.displayText());

return false;

}

return true;

const TablesStatusRequest & request)

{

if (!connected)

connect(timeouts);

TimeoutSetter timeout_setter(*socket, sync_request_timeout, true);

writeVarUInt(Protocol::Client::TablesStatusRequest, *out);

request.write(*out, server_revision);

out->next();

UInt64 response_type = 0;

readVarUInt(response_type, *in);

if (response_type == Protocol::Server::Exception)

receiveException()->rethrow();

else if (response_type != Protocol::Server::TablesStatusResponse)

throwUnexpectedPacket(response_type, "TablesStatusResponse");

TablesStatusResponse response;

response.read(*in, server_revision);

return response;

const ConnectionTimeouts & timeouts,

const String & query,

const String & query_id_,

UInt64 stage,

const Settings * settings,

const ClientInfo * client_info,

bool with_pending_data)

{

if (!connected)

connect(timeouts);

TimeoutSetter timeout_setter(*socket, timeouts.send_timeout, timeouts.receive_timeout, true);

if (settings)

{

std::optional<int> level;

std::string method = Poco::toUpper(settings->network_compression_method.toString());

/// Bad custom logic

if (method == "ZSTD")

level = settings->network_zstd_compression_level;

CompressionCodecFactory::instance().validateCodec(method, level, !settings->allow_suspicious_codecs);

compression_codec = CompressionCodecFactory::instance().get(method, level);

}

else

compression_codec = CompressionCodecFactory::instance().getDefaultCodec();

query_id = query_id_;

writeVarUInt(Protocol::Client::Query, *out);

writeStringBinary(query_id, *out);

/// Client info.

if (server_revision >= DBMS_MIN_REVISION_WITH_CLIENT_INFO)

{

if (client_info && !client_info->empty())

client_info->write(*out, server_revision);

else

ClientInfo().write(*out, server_revision);

}

/// Per query settings.

if (settings)

{

auto settings_format = (server_revision >= DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS) ? SettingsWriteFormat::STRINGS_WITH_FLAGS

: SettingsWriteFormat::BINARY;

settings->write(*out, settings_format);

}

else

writeStringBinary("" /* empty string is a marker of the end of settings */, *out);

/// Interserver secret

if (server_revision >= DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET)

{

/// Hash

///

/// Send correct hash only for !INITIAL_QUERY, due to:

/// - this will avoid extra protocol complexity for simplest cases

/// - there is no need in hash for the INITIAL_QUERY anyway

/// (since there is no secure/unsecure changes)

if (client_info && !cluster_secret.empty() && client_info->query_kind != ClientInfo::QueryKind::INITIAL_QUERY)

{

#if USE_SSL

std::string data(salt);

data += cluster_secret;

data += query;

data += query_id;

data += client_info->initial_user;

/// TODO: add source/target host/ip-address

std::string hash = encodeSHA256(data);

writeStringBinary(hash, *out);

#else

throw Exception(

"Inter-server secret support is disabled, because ClickHouse was built without SSL library",

ErrorCodes::SUPPORT_IS_DISABLED);

#endif

}

else

writeStringBinary("", *out);

}

writeVarUInt(stage, *out);

writeVarUInt(static_cast<bool>(compression), *out);

writeStringBinary(query, *out);

maybe_compressed_in.reset();

maybe_compressed_out.reset();

block_in.reset();

block_logs_in.reset();

block_out.reset();

/// Send empty block which means end of data.

if (!with_pending_data)

{

sendData(Block());

out->next();

}

{

/// If we already disconnected.

if (!out)

return;

writeVarUInt(Protocol::Client::Cancel, *out);

out->next();

{

if (!block_out)

{

if (compression == Protocol::Compression::Enable)

maybe_compressed_out = std::make_shared<CompressedWriteBuffer>(*out, compression_codec);

else

maybe_compressed_out = out;

block_out = std::make_shared<NativeBlockOutputStream>(*maybe_compressed_out, server_revision, block.cloneEmpty());

}

if (scalar)

writeVarUInt(Protocol::Client::Scalar, *out);

else

writeVarUInt(Protocol::Client::Data, *out);

writeStringBinary(name, *out);

size_t prev_bytes = out->count();

block_out->write(block);

maybe_compressed_out->next();

out->next();

if (throttler)

throttler->add(out->count() - prev_bytes);

{

/// NOTE 'Throttler' is not used in this method (could use, but it's not important right now).

writeVarUInt(Protocol::Client::Data, *out);

writeStringBinary(name, *out);

if (0 == size)

copyData(input, *out);

else

copyData(input, *out, size);

out->next();

{

if (data.empty())

return;

Stopwatch watch;

size_t out_bytes = out ? out->count() : 0;

size_t maybe_compressed_out_bytes = maybe_compressed_out ? maybe_compressed_out->count() : 0;

size_t rows = 0;

CurrentMetrics::Increment metric_increment{CurrentMetrics::SendScalars};

for (auto & elem : data)

{

rows += elem.second.rows();

sendData(elem.second, elem.first, true /* scalar */);

}

out_bytes = out->count() - out_bytes;

maybe_compressed_out_bytes = maybe_compressed_out->count() - maybe_compressed_out_bytes;

double elapsed = watch.elapsedSeconds();

if (compression == Protocol::Compression::Enable)

LOG_DEBUG(log_wrapper.get(),

"Sent data for {} scalars, total {} rows in {} sec., {} rows/sec., {} ({}/sec.), compressed {} times to {} ({}/sec.)",

data.size(), rows, elapsed,

static_cast<size_t>(rows / watch.elapsedSeconds()),

ReadableSize(maybe_compressed_out_bytes),

ReadableSize(maybe_compressed_out_bytes / watch.elapsedSeconds()),

static_cast<double>(maybe_compressed_out_bytes) / out_bytes,

ReadableSize(out_bytes),

ReadableSize(out_bytes / watch.elapsedSeconds()));

else

LOG_DEBUG(log_wrapper.get(),

"Sent data for {} scalars, total {} rows in {} sec., {} rows/sec., {} ({}/sec.), no compression.",

data.size(), rows, elapsed,

static_cast<size_t>(rows / watch.elapsedSeconds()),

ReadableSize(maybe_compressed_out_bytes),

ReadableSize(maybe_compressed_out_bytes / watch.elapsedSeconds()));

{

using OnCancell = std::function<void()>;

ExternalTableDataSink(Block header, Connection & connection_, ExternalTableData & table_data_, OnCancell callback)

: ISink(std::move(header)), connection(connection_), table_data(table_data_),

on_cancell(std::move(callback))

{}

String getName() const override { return "ExternalTableSink"; }

size_t getNumReadRows() const { return num_rows; }

void consume(Chunk chunk) override

{

if (table_data.is_cancelled)

{

on_cancell();

return;

}

num_rows += chunk.getNumRows();

auto block = getPort().getHeader().cloneWithColumns(chunk.detachColumns());

connection.sendData(block, table_data.table_name);

}

Connection & connection;

ExternalTableData & table_data;

OnCancell on_cancell;

size_t num_rows = 0;

{

if (data.empty())

{

/// Send empty block, which means end of data transfer.

sendData(Block());

return;

}

Stopwatch watch;

size_t out_bytes = out ? out->count() : 0;

size_t maybe_compressed_out_bytes = maybe_compressed_out ? maybe_compressed_out->count() : 0;

size_t rows = 0;

CurrentMetrics::Increment metric_increment{CurrentMetrics::SendExternalTables};

for (auto & elem : data)

{

PipelineExecutorPtr executor;

auto on_cancel = [& executor]() { executor->cancel(); };

QueryPipeline pipeline;

pipeline.init(std::move(*elem->pipe));

pipeline.resize(1);

auto sink = std::make_shared<ExternalTableDataSink>(pipeline.getHeader(), *this, *elem, std::move(on_cancel));

pipeline.setSinks([&](const Block &, QueryPipeline::StreamType type) -> ProcessorPtr

{

if (type != QueryPipeline::StreamType::Main)

return nullptr;

return sink;

});

executor = pipeline.execute();

executor->execute(/*num_threads = */ 1);

auto read_rows = sink->getNumReadRows();

rows += read_rows;

/// If table is empty, send empty block with name.

if (read_rows == 0)

sendData(sink->getPort().getHeader(), elem->table_name);

}

/// Send empty block, which means end of data transfer.

sendData(Block());

out_bytes = out->count() - out_bytes;

maybe_compressed_out_bytes = maybe_compressed_out->count() - maybe_compressed_out_bytes;

double elapsed = watch.elapsedSeconds();

if (compression == Protocol::Compression::Enable)

LOG_DEBUG(log_wrapper.get(),

"Sent data for {} external tables, total {} rows in {} sec., {} rows/sec., {} ({}/sec.), compressed {} times to {} ({}/sec.)",

data.size(), rows, elapsed,

static_cast<size_t>(rows / watch.elapsedSeconds()),

ReadableSize(maybe_compressed_out_bytes),

ReadableSize(maybe_compressed_out_bytes / watch.elapsedSeconds()),

static_cast<double>(maybe_compressed_out_bytes) / out_bytes,

ReadableSize(out_bytes),

ReadableSize(out_bytes / watch.elapsedSeconds()));

else

LOG_DEBUG(log_wrapper.get(),

"Sent data for {} external tables, total {} rows in {} sec., {} rows/sec., {} ({}/sec.), no compression.",

data.size(), rows, elapsed,

static_cast<size_t>(rows / watch.elapsedSeconds()),

ReadableSize(maybe_compressed_out_bytes),

ReadableSize(maybe_compressed_out_bytes / watch.elapsedSeconds()));

{

if (last_input_packet_type.has_value())

return last_input_packet_type;

if (hasReadPendingData() || poll(timeout_microseconds))

{

// LOG_TRACE(log_wrapper.get(), "Receiving packet type");

UInt64 packet_type;

readVarUInt(packet_type, *in);

last_input_packet_type.emplace(packet_type);

return last_input_packet_type;

}

return {};

{

try

{

Packet res;

/// Have we already read packet type?

if (last_input_packet_type)

{

res.type = *last_input_packet_type;

last_input_packet_type.reset();

}

else

{

readVarUInt(res.type, *in);

}

switch (res.type)

{

case Protocol::Server::Data: [[fallthrough]];

case Protocol::Server::Totals: [[fallthrough]];

case Protocol::Server::Extremes:

res.block = receiveData();

return res;

case Protocol::Server::Exception:

res.exception = receiveException();

return res;

case Protocol::Server::Progress:

res.progress = receiveProgress();

return res;

case Protocol::Server::ProfileInfo:

res.profile_info = receiveProfileInfo();

return res;

case Protocol::Server::Log:

res.block = receiveLogData();

return res;

case Protocol::Server::TableColumns:

res.multistring_message = receiveMultistringMessage(res.type);

return res;

case Protocol::Server::EndOfStream:

return res;

default:

/// In unknown state, disconnect - to not leave unsynchronised connection.

disconnect();

throw Exception("Unknown packet "

+ toString(res.type)

+ " from server " + getDescription(), ErrorCodes::UNKNOWN_PACKET_FROM_SERVER);

}

}

catch (Exception & e)

{

/// Add server address to exception message, if need.

if (e.code() != ErrorCodes::UNKNOWN_PACKET_FROM_SERVER)

e.addMessage("while receiving packet from " + getDescription());

throw;

}

{

initBlockInput();

return receiveDataImpl(block_in);

{

initBlockLogsInput();

return receiveDataImpl(block_logs_in);

{

String external_table_name;

readStringBinary(external_table_name, *in);

size_t prev_bytes = in->count();

/// Read one block from network.

Block res = stream->read();

if (throttler)

throttler->add(in->count() - prev_bytes);

return res;

{

if (!block_in)

{

if (!maybe_compressed_in)

{

if (compression == Protocol::Compression::Enable)

maybe_compressed_in = std::make_shared<CompressedReadBuffer>(*in);

else

maybe_compressed_in = in;

}

block_in = std::make_shared<NativeBlockInputStream>(*maybe_compressed_in, server_revision);

}

{

if (!block_logs_in)

{

/// Have to return superset of SystemLogsQueue::getSampleBlock() columns

block_logs_in = std::make_shared<NativeBlockInputStream>(*in, server_revision);

}

{

auto resolved_address = getResolvedAddress();

description = host + ":" + toString(port);

if (resolved_address)

{

auto ip_address = resolved_address->host().toString();

if (host != ip_address)

description += ", " + ip_address;

}

{

size_t num = Protocol::Server::stringsInMessage(msg_type);

std::vector<String> strings(num);

for (size_t i = 0; i < num; ++i)

readStringBinary(strings[i], *in);

return strings;

{

Progress progress;

progress.read(*in, server_revision);

return progress;

{

BlockStreamProfileInfo profile_info;

profile_info.read(*in);

return profile_info;

{

throw NetException(

"Unexpected packet from server " + getDescription() + " (expected " + expected

+ ", got " + String(Protocol::Server::toString(packet_type)) + ")",

ErrorCodes::UNEXPECTED_PACKET_FROM_SERVER);