ClickHouse/src/Server/TCPHandler.cpp

1810 lines
59 KiB
C++
Raw Normal View History

2012-03-19 12:57:56 +00:00
#include <iomanip>
2021-10-02 07:13:14 +00:00
#include <base/scope_guard.h>
2013-01-13 22:13:54 +00:00
#include <Poco/Net/NetException.h>
#include <Poco/Util/LayeredConfiguration.h>
#include <Common/CurrentThread.h>
#include <Common/Stopwatch.h>
#include <Common/NetException.h>
2018-08-31 00:59:48 +00:00
#include <Common/setThreadName.h>
#include <Common/OpenSSLHelpers.h>
#include <IO/Progress.h>
2018-12-28 18:15:26 +00:00
#include <Compression/CompressedReadBuffer.h>
#include <Compression/CompressedWriteBuffer.h>
#include <IO/ReadBufferFromPocoSocket.h>
#include <IO/WriteBufferFromPocoSocket.h>
2020-12-02 21:05:51 +00:00
#include <IO/LimitReadBuffer.h>
2018-09-20 20:51:21 +00:00
#include <IO/ReadHelpers.h>
#include <IO/WriteHelpers.h>
#include <IO/copyData.h>
2021-10-08 17:21:19 +00:00
#include <DataStreams/NativeReader.h>
#include <DataStreams/NativeWriter.h>
#include <Interpreters/executeQuery.h>
#include <Interpreters/TablesStatus.h>
#include <Interpreters/InternalTextLogsQueue.h>
#include <Interpreters/OpenTelemetrySpanLog.h>
2021-08-01 14:12:34 +00:00
#include <Interpreters/Session.h>
#include <Storages/StorageReplicatedMergeTree.h>
#include <Storages/MergeTree/MergeTreeDataPartUUID.h>
2021-04-12 21:42:52 +00:00
#include <Storages/StorageS3Cluster.h>
#include <Core/ExternalTable.h>
2021-08-01 14:12:34 +00:00
#include <Access/Credentials.h>
#include <Storages/ColumnDefault.h>
#include <DataTypes/DataTypeLowCardinality.h>
2021-08-30 11:04:59 +00:00
#include <DataTypes/DataTypeEnum.h>
2018-12-21 12:17:30 +00:00
#include <Compression/CompressionFactory.h>
2021-10-02 07:13:14 +00:00
#include <base/logger_useful.h>
#include <Common/CurrentMetrics.h>
#include <fmt/format.h>
2020-05-20 19:01:36 +00:00
#include <Processors/Executors/PullingAsyncPipelineExecutor.h>
#include <Processors/Executors/PushingPipelineExecutor.h>
2021-09-10 14:52:24 +00:00
#include <Processors/Executors/PushingAsyncPipelineExecutor.h>
2021-09-15 19:35:48 +00:00
#include <Processors/Executors/CompletedPipelineExecutor.h>
#include <Processors/Sinks/SinkToStorage.h>
2019-03-26 18:28:37 +00:00
2021-03-16 18:41:29 +00:00
#include "Core/Protocol.h"
2012-03-09 15:46:52 +00:00
#include "TCPHandler.h"
#if !defined(ARCADIA_BUILD)
# include <Common/config_version.h>
#endif
namespace CurrentMetrics
{
extern const Metric QueryThread;
}
2012-03-09 15:46:52 +00:00
namespace DB
{
namespace ErrorCodes
{
2020-02-25 18:02:41 +00:00
extern const int LOGICAL_ERROR;
extern const int ATTEMPT_TO_READ_AFTER_EOF;
extern const int CLIENT_HAS_CONNECTED_TO_WRONG_PORT;
extern const int UNKNOWN_EXCEPTION;
extern const int UNKNOWN_PACKET_FROM_CLIENT;
extern const int POCO_EXCEPTION;
extern const int SOCKET_TIMEOUT;
extern const int UNEXPECTED_PACKET_FROM_CLIENT;
extern const int SUPPORT_IS_DISABLED;
2021-04-12 17:48:16 +00:00
extern const int UNKNOWN_PROTOCOL;
}
TCPHandler::TCPHandler(IServer & server_, const Poco::Net::StreamSocket & socket_, bool parse_proxy_protocol_, std::string server_display_name_)
: Poco::Net::TCPServerConnection(socket_)
, server(server_)
, parse_proxy_protocol(parse_proxy_protocol_)
, log(&Poco::Logger::get("TCPHandler"))
, server_display_name(std::move(server_display_name_))
{
}
TCPHandler: catch exceptions from the WriteBuffer in destructor For TCPHandler it is safe thing todo. Otherwise *San will report [1]: 2021.01.24 15:33:40.103996 [ 270 ] {} <Trace> BaseDaemon: Received signal -1 2021.01.24 15:33:40.110693 [ 270 ] {} <Fatal> BaseDaemon: (version 21.2.1.5789, build id: FF421B087D1E2EAA19FA17B5AB3AE413832744E0) (from thread 48318) Terminate called for uncaught exception: 2021.01.24 15:33:40.114845 [ 270 ] {} <Trace> BaseDaemon: Received signal 6 2021.01.24 15:33:40.138738 [ 218027 ] {} <Fatal> BaseDaemon: ######################################## 2021.01.24 15:33:40.138838 [ 218027 ] {} <Fatal> BaseDaemon: (version 21.2.1.5789, build id: FF421B087D1E2EAA19FA17B5AB3AE413832744E0) (from thread 48318) (no query) Received signal Aborted (6) 2021.01.24 15:33:40.138912 [ 218027 ] {} <Fatal> BaseDaemon: 2021.01.24 15:33:40.139277 [ 218027 ] {} <Fatal> BaseDaemon: Stack trace: 0x7f185474118b 0x7f1854720859 0xaddc0cc 0x2af9fab8 0x2af9fa04 0xa91758b 0x1e418bb5 0x20725b4f 0x20725d9e 0x266b47a3 0x269772f5 0x26971847 0x7f18548f6609 0x7f185481d293 2021.01.24 15:33:40.139637 [ 218027 ] {} <Fatal> BaseDaemon: 3. raise @ 0x4618b in /usr/lib/x86_64-linux-gnu/libc-2.31.so 2021.01.24 15:33:40.140113 [ 218027 ] {} <Fatal> BaseDaemon: 4. abort @ 0x25859 in /usr/lib/x86_64-linux-gnu/libc-2.31.so 2021.01.24 15:33:40.144121 [ 218027 ] {} <Fatal> BaseDaemon: 5. ./obj-x86_64-linux-gnu/../base/daemon/BaseDaemon.cpp:0: terminate_handler() @ 0xaddc0cc in /usr/bin/clickhouse 2021.01.24 15:33:40.151208 [ 218027 ] {} <Fatal> BaseDaemon: 6. ./obj-x86_64-linux-gnu/../contrib/libcxxabi/src/cxa_handlers.cpp:61: std::__terminate(void (*)()) @ 0x2af9fab8 in /usr/bin/clickhouse 2021.01.24 15:33:40.153085 [ 218027 ] {} <Fatal> BaseDaemon: 7. ./obj-x86_64-linux-gnu/../contrib/libcxxabi/src/cxa_handlers.cpp:0: std::terminate() @ 0x2af9fa04 in /usr/bin/clickhouse 2021.01.24 15:33:40.155209 [ 218027 ] {} <Fatal> BaseDaemon: 8. ? @ 0xa91758b in /usr/bin/clickhouse 2021.01.24 15:33:40.156621 [ 218027 ] {} <Fatal> BaseDaemon: 9. ./obj-x86_64-linux-gnu/../src/IO/WriteBufferFromPocoSocket.cpp:0: DB::WriteBufferFromPocoSocket::~WriteBufferFromPocoSocket() @ 0x1e418bb5 in /usr/bin/clickhouse 2021.01.24 15:33:40.161041 [ 218027 ] {} <Fatal> BaseDaemon: 10. ./obj-x86_64-linux-gnu/../contrib/libcxx/include/memory:2518: DB::TCPHandler::~TCPHandler() @ 0x20725b4f in /usr/bin/clickhouse 2021.01.24 15:33:40.164557 [ 218027 ] {} <Fatal> BaseDaemon: 11. ./obj-x86_64-linux-gnu/../src/Server/TCPHandler.h:101: DB::TCPHandler::~TCPHandler() @ 0x20725d9e in /usr/bin/clickhouse 2021.01.24 15:33:40.165921 [ 218027 ] {} <Fatal> BaseDaemon: 12. ./obj-x86_64-linux-gnu/../contrib/poco/Foundation/include/Poco/AtomicCounter.h:314: Poco::Net::TCPServerDispatcher::run() @ 0x266b47a3 in /usr/bin/clickhouse 2021.01.24 15:33:40.167347 [ 218027 ] {} <Fatal> BaseDaemon: 13. ./obj-x86_64-linux-gnu/../contrib/poco/Foundation/src/ThreadPool.cpp:0: Poco::PooledThread::run() @ 0x269772f5 in /usr/bin/clickhouse 2021.01.24 15:33:40.169401 [ 218027 ] {} <Fatal> BaseDaemon: 14. ./obj-x86_64-linux-gnu/../contrib/poco/Foundation/src/Thread_POSIX.cpp:0: Poco::ThreadImpl::runnableEntry(void*) @ 0x26971847 in /usr/bin/clickhouse 2021.01.24 15:33:40.169498 [ 218027 ] {} <Fatal> BaseDaemon: 15. start_thread @ 0x9609 in /usr/lib/x86_64-linux-gnu/libpthread-2.31.so 2021.01.24 15:33:40.169566 [ 218027 ] {} <Fatal> BaseDaemon: 16. __clone @ 0x122293 in /usr/lib/x86_64-linux-gnu/libc-2.31.so 2021.01.24 15:33:41.027601 [ 218027 ] {} <Fatal> BaseDaemon: Calculated checksum of the binary: 63D7491B39260494BA0D785E1860B427. There is no information about the reference checksum. [1]: https://clickhouse-test-reports.s3.yandex.net/19451/1e16bd6f337985a82fbdf4eded695dc6e663af58/stress_test_(address).html#fail1 v2: Fix catching errors in WriteBufferFromPocoSocket destructor
2021-01-28 04:07:51 +00:00
TCPHandler::~TCPHandler()
{
try
{
state.reset();
if (out)
out->next();
TCPHandler: catch exceptions from the WriteBuffer in destructor For TCPHandler it is safe thing todo. Otherwise *San will report [1]: 2021.01.24 15:33:40.103996 [ 270 ] {} <Trace> BaseDaemon: Received signal -1 2021.01.24 15:33:40.110693 [ 270 ] {} <Fatal> BaseDaemon: (version 21.2.1.5789, build id: FF421B087D1E2EAA19FA17B5AB3AE413832744E0) (from thread 48318) Terminate called for uncaught exception: 2021.01.24 15:33:40.114845 [ 270 ] {} <Trace> BaseDaemon: Received signal 6 2021.01.24 15:33:40.138738 [ 218027 ] {} <Fatal> BaseDaemon: ######################################## 2021.01.24 15:33:40.138838 [ 218027 ] {} <Fatal> BaseDaemon: (version 21.2.1.5789, build id: FF421B087D1E2EAA19FA17B5AB3AE413832744E0) (from thread 48318) (no query) Received signal Aborted (6) 2021.01.24 15:33:40.138912 [ 218027 ] {} <Fatal> BaseDaemon: 2021.01.24 15:33:40.139277 [ 218027 ] {} <Fatal> BaseDaemon: Stack trace: 0x7f185474118b 0x7f1854720859 0xaddc0cc 0x2af9fab8 0x2af9fa04 0xa91758b 0x1e418bb5 0x20725b4f 0x20725d9e 0x266b47a3 0x269772f5 0x26971847 0x7f18548f6609 0x7f185481d293 2021.01.24 15:33:40.139637 [ 218027 ] {} <Fatal> BaseDaemon: 3. raise @ 0x4618b in /usr/lib/x86_64-linux-gnu/libc-2.31.so 2021.01.24 15:33:40.140113 [ 218027 ] {} <Fatal> BaseDaemon: 4. abort @ 0x25859 in /usr/lib/x86_64-linux-gnu/libc-2.31.so 2021.01.24 15:33:40.144121 [ 218027 ] {} <Fatal> BaseDaemon: 5. ./obj-x86_64-linux-gnu/../base/daemon/BaseDaemon.cpp:0: terminate_handler() @ 0xaddc0cc in /usr/bin/clickhouse 2021.01.24 15:33:40.151208 [ 218027 ] {} <Fatal> BaseDaemon: 6. ./obj-x86_64-linux-gnu/../contrib/libcxxabi/src/cxa_handlers.cpp:61: std::__terminate(void (*)()) @ 0x2af9fab8 in /usr/bin/clickhouse 2021.01.24 15:33:40.153085 [ 218027 ] {} <Fatal> BaseDaemon: 7. ./obj-x86_64-linux-gnu/../contrib/libcxxabi/src/cxa_handlers.cpp:0: std::terminate() @ 0x2af9fa04 in /usr/bin/clickhouse 2021.01.24 15:33:40.155209 [ 218027 ] {} <Fatal> BaseDaemon: 8. ? @ 0xa91758b in /usr/bin/clickhouse 2021.01.24 15:33:40.156621 [ 218027 ] {} <Fatal> BaseDaemon: 9. ./obj-x86_64-linux-gnu/../src/IO/WriteBufferFromPocoSocket.cpp:0: DB::WriteBufferFromPocoSocket::~WriteBufferFromPocoSocket() @ 0x1e418bb5 in /usr/bin/clickhouse 2021.01.24 15:33:40.161041 [ 218027 ] {} <Fatal> BaseDaemon: 10. ./obj-x86_64-linux-gnu/../contrib/libcxx/include/memory:2518: DB::TCPHandler::~TCPHandler() @ 0x20725b4f in /usr/bin/clickhouse 2021.01.24 15:33:40.164557 [ 218027 ] {} <Fatal> BaseDaemon: 11. ./obj-x86_64-linux-gnu/../src/Server/TCPHandler.h:101: DB::TCPHandler::~TCPHandler() @ 0x20725d9e in /usr/bin/clickhouse 2021.01.24 15:33:40.165921 [ 218027 ] {} <Fatal> BaseDaemon: 12. ./obj-x86_64-linux-gnu/../contrib/poco/Foundation/include/Poco/AtomicCounter.h:314: Poco::Net::TCPServerDispatcher::run() @ 0x266b47a3 in /usr/bin/clickhouse 2021.01.24 15:33:40.167347 [ 218027 ] {} <Fatal> BaseDaemon: 13. ./obj-x86_64-linux-gnu/../contrib/poco/Foundation/src/ThreadPool.cpp:0: Poco::PooledThread::run() @ 0x269772f5 in /usr/bin/clickhouse 2021.01.24 15:33:40.169401 [ 218027 ] {} <Fatal> BaseDaemon: 14. ./obj-x86_64-linux-gnu/../contrib/poco/Foundation/src/Thread_POSIX.cpp:0: Poco::ThreadImpl::runnableEntry(void*) @ 0x26971847 in /usr/bin/clickhouse 2021.01.24 15:33:40.169498 [ 218027 ] {} <Fatal> BaseDaemon: 15. start_thread @ 0x9609 in /usr/lib/x86_64-linux-gnu/libpthread-2.31.so 2021.01.24 15:33:40.169566 [ 218027 ] {} <Fatal> BaseDaemon: 16. __clone @ 0x122293 in /usr/lib/x86_64-linux-gnu/libc-2.31.so 2021.01.24 15:33:41.027601 [ 218027 ] {} <Fatal> BaseDaemon: Calculated checksum of the binary: 63D7491B39260494BA0D785E1860B427. There is no information about the reference checksum. [1]: https://clickhouse-test-reports.s3.yandex.net/19451/1e16bd6f337985a82fbdf4eded695dc6e663af58/stress_test_(address).html#fail1 v2: Fix catching errors in WriteBufferFromPocoSocket destructor
2021-01-28 04:07:51 +00:00
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__);
}
}
2012-03-09 15:46:52 +00:00
void TCPHandler::runImpl()
{
2018-08-31 00:59:48 +00:00
setThreadName("TCPHandler");
ThreadStatus thread_status;
2018-08-31 00:59:48 +00:00
session = std::make_unique<Session>(server.context(), ClientInfo::Interface::TCP);
2021-08-01 14:12:34 +00:00
extractConnectionSettingsFromContext(server.context());
2021-08-01 14:12:34 +00:00
socket().setReceiveTimeout(receive_timeout);
socket().setSendTimeout(send_timeout);
socket().setNoDelay(true);
in = std::make_shared<ReadBufferFromPocoSocket>(socket());
out = std::make_shared<WriteBufferFromPocoSocket>(socket());
2020-12-02 21:05:51 +00:00
/// Support for PROXY protocol
if (parse_proxy_protocol && !receiveProxyHeader())
return;
if (in->eof())
{
LOG_INFO(log, "Client has not sent any data.");
return;
}
/// User will be authenticated here. It will also set settings from user profile into connection_context.
try
{
receiveHello();
sendHello();
if (!is_interserver_mode) /// In interserver mode queries are executed without a session context.
{
session->makeSessionContext();
/// If session created, then settings in session context has been updated.
/// So it's better to update the connection settings for flexibility.
extractConnectionSettingsFromContext(session->sessionContext());
/// When connecting, the default database could be specified.
if (!default_database.empty())
session->sessionContext()->setCurrentDatabase(default_database);
}
}
catch (const Exception & e) /// Typical for an incorrect username, password, or address.
{
if (e.code() == ErrorCodes::CLIENT_HAS_CONNECTED_TO_WRONG_PORT)
{
LOG_DEBUG(log, "Client has connected to wrong port.");
return;
}
if (e.code() == ErrorCodes::ATTEMPT_TO_READ_AFTER_EOF)
{
LOG_INFO(log, "Client has gone away.");
return;
}
try
{
/// We try to send error information to the client.
2021-08-01 14:12:34 +00:00
sendException(e, send_exception_with_stack_trace);
}
catch (...) {}
throw;
}
2020-03-08 21:29:00 +00:00
while (true)
{
2019-07-04 22:57:26 +00:00
/// We are waiting for a packet from the client. Thus, every `poll_interval` seconds check whether we need to shut down.
2019-07-04 22:23:45 +00:00
{
Stopwatch idle_time;
UInt64 timeout_ms = std::min(poll_interval, idle_connection_timeout) * 1000000;
while (!server.isCancelled() && !static_cast<ReadBufferFromPocoSocket &>(*in).poll(timeout_ms))
2019-07-04 22:23:45 +00:00
{
if (idle_time.elapsedSeconds() > idle_connection_timeout)
2019-07-04 23:03:20 +00:00
{
LOG_TRACE(log, "Closing idle connection");
2019-07-04 22:23:45 +00:00
return;
2019-07-04 23:03:20 +00:00
}
2019-07-04 22:23:45 +00:00
}
}
/// If we need to shut down, or client disconnects.
2017-08-09 14:33:07 +00:00
if (server.isCancelled() || in->eof())
break;
Stopwatch watch;
state.reset();
/// Initialized later.
std::optional<CurrentThread::QueryScope> query_scope;
/** An exception during the execution of request (it must be sent over the network to the client).
* The client will be able to accept it, if it did not happen while sending another packet and the client has not disconnected yet.
*/
std::optional<DB::Exception> exception;
bool network_error = false;
try
{
/// If a user passed query-local timeouts, reset socket to initial state at the end of the query
SCOPE_EXIT({state.timeout_setter.reset();});
/** If Query - process it. If Ping or Cancel - go back to the beginning.
* There may come settings for a separate query that modify `query_context`.
* It's possible to receive part uuids packet before the query, so then receivePacket has to be called twice.
*/
if (!receivePacket())
continue;
/** If part_uuids got received in previous packet, trying to read again.
*/
2021-08-01 14:12:34 +00:00
if (state.empty() && state.part_uuids_to_ignore && !receivePacket())
continue;
query_scope.emplace(query_context);
2021-08-01 14:12:34 +00:00
/// If query received, then settings in query_context has been updated.
/// So it's better to update the connection settings for flexibility.
extractConnectionSettingsFromContext(query_context);
/// Sync timeouts on client and server during current query to avoid dangling queries on server
/// NOTE: We use send_timeout for the receive timeout and vice versa (change arguments ordering in TimeoutSetter),
/// because send_timeout is client-side setting which has opposite meaning on the server side.
/// NOTE: these settings are applied only for current connection (not for distributed tables' connections)
state.timeout_setter = std::make_unique<TimeoutSetter>(socket(), receive_timeout, send_timeout);
2018-08-24 07:30:53 +00:00
std::mutex fatal_error_mutex;
/// Should we send internal logs to client?
2019-07-10 12:19:17 +00:00
const auto client_logs_level = query_context->getSettingsRef().send_logs_level;
2020-09-17 12:15:05 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_SERVER_LOGS
2019-08-09 13:02:01 +00:00
&& client_logs_level != LogsLevel::none)
{
state.logs_queue = std::make_shared<InternalTextLogsQueue>();
2019-07-10 12:19:17 +00:00
state.logs_queue->max_priority = Poco::Logger::parseLevel(client_logs_level.toString());
2019-08-09 13:02:01 +00:00
CurrentThread::attachInternalTextLogsQueue(state.logs_queue, client_logs_level);
CurrentThread::setFatalErrorCallback([this, &fatal_error_mutex]
{
std::lock_guard lock(fatal_error_mutex);
sendLogs();
});
}
2021-08-01 14:12:34 +00:00
query_context->setExternalTablesInitializer([this] (ContextPtr context)
2018-08-24 00:07:25 +00:00
{
if (context != query_context)
throw Exception("Unexpected context in external tables initializer", ErrorCodes::LOGICAL_ERROR);
/// Get blocks of temporary tables
2021-08-01 14:12:34 +00:00
readData();
/// Reset the input stream, as we received an empty block while receiving external table data.
/// So, the stream has been marked as cancelled and we can't read from it anymore.
state.block_in.reset();
2018-11-26 00:56:50 +00:00
state.maybe_compressed_in.reset(); /// For more accurate accounting by MemoryTracker.
});
2019-05-28 18:30:10 +00:00
/// Send structure of columns to client for function input()
query_context->setInputInitializer([this] (ContextPtr context, const StoragePtr & input_storage)
2019-05-28 18:30:10 +00:00
{
if (context != query_context)
2019-05-28 18:30:10 +00:00
throw Exception("Unexpected context in Input initializer", ErrorCodes::LOGICAL_ERROR);
auto metadata_snapshot = input_storage->getInMemoryMetadataPtr();
2019-05-28 18:30:10 +00:00
state.need_receive_data_for_input = true;
/// Send ColumnsDescription for input storage.
2020-09-17 12:15:05 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_COLUMN_DEFAULTS_METADATA
2019-05-28 18:30:10 +00:00
&& query_context->getSettingsRef().input_format_defaults_for_omitted_fields)
{
sendTableColumns(metadata_snapshot->getColumns());
2019-05-28 18:30:10 +00:00
}
/// Send block to the client - input storage structure.
state.input_header = metadata_snapshot->getSampleBlock();
2019-05-28 18:30:10 +00:00
sendData(state.input_header);
});
2021-08-01 14:12:34 +00:00
query_context->setInputBlocksReaderCallback([this] (ContextPtr context) -> Block
2019-05-28 18:30:10 +00:00
{
if (context != query_context)
2019-05-28 18:30:10 +00:00
throw Exception("Unexpected context in InputBlocksReader", ErrorCodes::LOGICAL_ERROR);
2021-08-01 14:12:34 +00:00
if (!readDataNext())
2019-05-28 18:30:10 +00:00
{
state.block_in.reset();
state.maybe_compressed_in.reset();
return Block();
2019-05-28 20:16:24 +00:00
}
2019-05-28 18:30:10 +00:00
return state.block_for_input;
});
customizeContext(query_context);
2021-04-12 17:07:01 +00:00
/// This callback is needed for requesting read tasks inside pipeline for distributed processing
2021-04-13 10:59:02 +00:00
query_context->setReadTaskCallback([this]() -> String
{
2021-04-08 19:00:39 +00:00
std::lock_guard lock(task_callback_mutex);
2021-04-10 02:21:18 +00:00
sendReadTaskRequestAssumeLocked();
return receiveReadTaskResponseAssumeLocked();
});
/// Processing Query
state.io = executeQuery(state.query, query_context, false, state.stage);
after_check_cancelled.restart();
after_send_progress.restart();
2021-09-15 19:35:48 +00:00
if (state.io.pipeline.pushing())
/// FIXME: check explicitly that insert query suggests to receive data via native protocol,
2020-10-10 01:43:07 +00:00
{
state.need_receive_data_for_insert = true;
2021-08-01 14:12:34 +00:00
processInsertQuery();
2020-10-10 01:43:07 +00:00
}
2021-09-16 17:40:42 +00:00
else if (state.io.pipeline.pulling())
{
processOrdinaryQueryWithProcessors();
}
2021-09-17 11:40:03 +00:00
else if (state.io.pipeline.completed())
2019-05-28 18:30:10 +00:00
{
2021-09-15 19:35:48 +00:00
CompletedPipelineExecutor executor(state.io.pipeline);
2021-09-22 10:57:00 +00:00
/// Should not check for cancel in case of input.
if (!state.need_receive_data_for_input)
2021-09-22 13:29:58 +00:00
{
auto callback = [this, &fatal_error_mutex]()
2021-09-22 13:29:58 +00:00
{
std::lock_guard lock(fatal_error_mutex);
2021-09-22 13:29:58 +00:00
if (isQueryCancelled())
return true;
sendProgress();
sendLogs();
return false;
};
executor.setCancelCallback(callback, interactive_delay / 1000);
}
2021-09-15 19:35:48 +00:00
executor.execute();
2019-05-28 18:30:10 +00:00
}
2020-10-10 09:24:12 +00:00
state.io.onFinish();
/// Do it before sending end of stream, to have a chance to show log message in client.
query_scope->logPeakMemoryUsage();
if (state.is_connection_closed)
break;
sendLogs();
sendEndOfStream();
Fix memory tracking for two-level GROUP BY when not all rows read from Aggregator (TCP) Example of such cases: - SELECT GROUP BY LIMIT - SELECT GROUP BY with subsequent MEMORY_LIMIT_EXCEEDED error And it should be two-level aggregation, since otherwise there will be only one hashtable which will be cleared correctly, only if you have two-level GROUP BY some of hashtables will not be cleared since nobody consume rows. Before this patch: 09:39.015292 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.707 sec., 1413826 rows/sec., 10.79 MiB/sec. 09:39.015348 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.015942 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- **problem** 09:39.017057 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- 09:39.017961 [ 3070801 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.018029 [ 3070801 ] {} <Information> TCPHandler: Processed in 0.711 sec. After this patch: 16:24.544030 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.599 sec., 1670199 rows/sec., 12.74 MiB/sec. 16:24.544084 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.544398 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.545485 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.547053 [ 3087333 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.547093 [ 3087333 ] {} <Information> TCPHandler: Processed in 0.603 sec.
2020-05-18 18:13:56 +00:00
/// QueryState should be cleared before QueryScope, since otherwise
/// the MemoryTracker will be wrong for possible deallocations.
/// (i.e. deallocations from the Aggregator with two-level aggregation)
state.reset();
Fix memory tracking for two-level GROUP BY when not all rows read from Aggregator (TCP) Example of such cases: - SELECT GROUP BY LIMIT - SELECT GROUP BY with subsequent MEMORY_LIMIT_EXCEEDED error And it should be two-level aggregation, since otherwise there will be only one hashtable which will be cleared correctly, only if you have two-level GROUP BY some of hashtables will not be cleared since nobody consume rows. Before this patch: 09:39.015292 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.707 sec., 1413826 rows/sec., 10.79 MiB/sec. 09:39.015348 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.015942 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- **problem** 09:39.017057 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- 09:39.017961 [ 3070801 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.018029 [ 3070801 ] {} <Information> TCPHandler: Processed in 0.711 sec. After this patch: 16:24.544030 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.599 sec., 1670199 rows/sec., 12.74 MiB/sec. 16:24.544084 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.544398 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.545485 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.547053 [ 3087333 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.547093 [ 3087333 ] {} <Information> TCPHandler: Processed in 0.603 sec.
2020-05-18 18:13:56 +00:00
query_scope.reset();
}
catch (const Exception & e)
{
state.io.onException();
exception.emplace(e);
if (e.code() == ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT)
throw;
/// If there is UNEXPECTED_PACKET_FROM_CLIENT emulate network_error
/// to break the loop, but do not throw to send the exception to
/// the client.
if (e.code() == ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT)
network_error = true;
/// If a timeout occurred, try to inform client about it and close the session
if (e.code() == ErrorCodes::SOCKET_TIMEOUT)
network_error = true;
}
catch (const Poco::Net::NetException & e)
{
/** We can get here if there was an error during connection to the client,
* or in connection with a remote server that was used to process the request.
* It is not possible to distinguish between these two cases.
* Although in one of them, we have to send exception to the client, but in the other - we can not.
* We will try to send exception to the client in any case - see below.
*/
state.io.onException();
exception.emplace(Exception::CreateFromPocoTag{}, e);
}
catch (const Poco::Exception & e)
{
state.io.onException();
exception.emplace(Exception::CreateFromPocoTag{}, e);
}
2020-07-17 14:25:13 +00:00
// Server should die on std logic errors in debug, like with assert()
// or ErrorCodes::LOGICAL_ERROR. This helps catch these errors in
// tests.
#ifndef NDEBUG
catch (const std::logic_error & e)
{
2020-07-17 14:20:33 +00:00
state.io.onException();
exception.emplace(Exception::CreateFromSTDTag{}, e);
sendException(*exception, send_exception_with_stack_trace);
2020-07-17 14:25:13 +00:00
std::abort();
}
2020-07-17 14:25:13 +00:00
#endif
catch (const std::exception & e)
{
state.io.onException();
exception.emplace(Exception::CreateFromSTDTag{}, e);
}
catch (...)
{
state.io.onException();
exception.emplace("Unknown exception", ErrorCodes::UNKNOWN_EXCEPTION);
}
try
{
if (exception)
{
try
{
/// Try to send logs to client, but it could be risky too
/// Assume that we can't break output here
sendLogs();
}
catch (...)
{
tryLogCurrentException(log, "Can't send logs to client");
}
const auto & e = *exception;
LOG_ERROR(log, getExceptionMessage(e, true));
2018-08-24 07:30:53 +00:00
sendException(*exception, send_exception_with_stack_trace);
}
}
catch (...)
{
/** Could not send exception information to the client. */
network_error = true;
LOG_WARNING(log, "Client has gone away.");
}
try
{
2021-08-01 14:12:34 +00:00
/// A query packet is always followed by one or more data packets.
/// If some of those data packets are left, try to skip them.
if (exception && !state.empty() && !state.read_all_data)
skipData();
}
catch (...)
{
network_error = true;
2021-08-01 14:12:34 +00:00
LOG_WARNING(log, "Can't skip data packets after query failure.");
}
try
{
Fix memory tracking for two-level GROUP BY when not all rows read from Aggregator (TCP) Example of such cases: - SELECT GROUP BY LIMIT - SELECT GROUP BY with subsequent MEMORY_LIMIT_EXCEEDED error And it should be two-level aggregation, since otherwise there will be only one hashtable which will be cleared correctly, only if you have two-level GROUP BY some of hashtables will not be cleared since nobody consume rows. Before this patch: 09:39.015292 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.707 sec., 1413826 rows/sec., 10.79 MiB/sec. 09:39.015348 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.015942 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- **problem** 09:39.017057 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- 09:39.017961 [ 3070801 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.018029 [ 3070801 ] {} <Information> TCPHandler: Processed in 0.711 sec. After this patch: 16:24.544030 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.599 sec., 1670199 rows/sec., 12.74 MiB/sec. 16:24.544084 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.544398 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.545485 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.547053 [ 3087333 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.547093 [ 3087333 ] {} <Information> TCPHandler: Processed in 0.603 sec.
2020-05-18 18:13:56 +00:00
/// QueryState should be cleared before QueryScope, since otherwise
/// the MemoryTracker will be wrong for possible deallocations.
/// (i.e. deallocations from the Aggregator with two-level aggregation)
state.reset();
Fix memory tracking for two-level GROUP BY when not all rows read from Aggregator (TCP) Example of such cases: - SELECT GROUP BY LIMIT - SELECT GROUP BY with subsequent MEMORY_LIMIT_EXCEEDED error And it should be two-level aggregation, since otherwise there will be only one hashtable which will be cleared correctly, only if you have two-level GROUP BY some of hashtables will not be cleared since nobody consume rows. Before this patch: 09:39.015292 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.707 sec., 1413826 rows/sec., 10.79 MiB/sec. 09:39.015348 [ 3070801 ] {609a0610-e377-4132-9cf3-f49454cf3c96} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.015942 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- **problem** 09:39.017057 [ 3070801 ] {} <Trace> Aggregator: Destroying aggregate states <-- 09:39.017961 [ 3070801 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 51.93 MiB. 09:39.018029 [ 3070801 ] {} <Information> TCPHandler: Processed in 0.711 sec. After this patch: 16:24.544030 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Information> executeQuery: Read 1000000 rows, 7.63 MiB in 0.599 sec., 1670199 rows/sec., 12.74 MiB/sec. 16:24.544084 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.544398 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.545485 [ 3087333 ] {79da208a-b3c0-48d4-9943-c974a3cbb6ea} <Trace> Aggregator: Destroying aggregate states 16:24.547053 [ 3087333 ] {} <Debug> MemoryTracker: Peak memory usage (for query): 72.11 MiB. 16:24.547093 [ 3087333 ] {} <Information> TCPHandler: Processed in 0.603 sec.
2020-05-18 18:13:56 +00:00
query_scope.reset();
}
catch (...)
{
/** During the processing of request, there was an exception that we caught and possibly sent to client.
* When destroying the request pipeline execution there was a second exception.
* For example, a pipeline could run in multiple threads, and an exception could occur in each of them.
* Ignore it.
*/
}
watch.stop();
2020-10-10 17:47:34 +00:00
LOG_DEBUG(log, "Processed in {} sec.", watch.elapsedSeconds());
/// It is important to destroy query context here. We do not want it to live arbitrarily longer than the query.
query_context.reset();
if (network_error)
break;
}
2012-03-11 08:52:56 +00:00
}
2021-08-01 14:12:34 +00:00
void TCPHandler::extractConnectionSettingsFromContext(const ContextPtr & context)
{
const auto & settings = context->getSettingsRef();
send_exception_with_stack_trace = settings.calculate_text_stack_trace;
send_timeout = settings.send_timeout;
receive_timeout = settings.receive_timeout;
poll_interval = settings.poll_interval;
idle_connection_timeout = settings.idle_connection_timeout;
interactive_delay = settings.interactive_delay;
sleep_in_send_tables_status = settings.sleep_in_send_tables_status_ms;
unknown_packet_in_send_data = settings.unknown_packet_in_send_data;
sleep_in_receive_cancel = settings.sleep_in_receive_cancel_ms;
}
bool TCPHandler::readDataNext()
2019-05-28 18:30:10 +00:00
{
Stopwatch watch(CLOCK_MONOTONIC_COARSE);
2021-08-01 14:12:34 +00:00
/// Poll interval should not be greater than receive_timeout
constexpr UInt64 min_timeout_ms = 5000; // 5 ms
UInt64 timeout_ms = std::max(min_timeout_ms, std::min(poll_interval * 1000000, static_cast<UInt64>(receive_timeout.totalMicroseconds())));
bool read_ok = false;
2019-05-28 18:30:10 +00:00
/// We are waiting for a packet from the client. Thus, every `POLL_INTERVAL` seconds check whether we need to shut down.
while (true)
{
2021-08-01 14:12:34 +00:00
if (static_cast<ReadBufferFromPocoSocket &>(*in).poll(timeout_ms))
{
/// If client disconnected.
if (in->eof())
{
LOG_INFO(log, "Client has dropped the connection, cancel the query.");
state.is_connection_closed = true;
break;
}
/// We accept and process data.
read_ok = receivePacket();
2021-04-06 19:18:45 +00:00
break;
2021-08-01 14:12:34 +00:00
}
2019-05-28 18:30:10 +00:00
/// Do we need to shut down?
if (server.isCancelled())
2021-08-01 14:12:34 +00:00
break;
2019-05-28 18:30:10 +00:00
/** Have we waited for data for too long?
* If we periodically poll, the receive_timeout of the socket itself does not work.
* Therefore, an additional check is added.
*/
2021-04-29 16:11:20 +00:00
Float64 elapsed = watch.elapsedSeconds();
2021-08-01 14:12:34 +00:00
if (elapsed > static_cast<Float64>(receive_timeout.totalSeconds()))
2019-05-28 18:30:10 +00:00
{
2020-11-10 18:22:26 +00:00
throw Exception(ErrorCodes::SOCKET_TIMEOUT,
"Timeout exceeded while receiving data from client. Waited for {} seconds, timeout is {} seconds.",
2021-08-01 14:12:34 +00:00
static_cast<size_t>(elapsed), receive_timeout.totalSeconds());
2019-05-28 18:30:10 +00:00
}
}
2021-08-01 14:12:34 +00:00
if (read_ok)
sendLogs();
else
state.read_all_data = true;
2019-05-28 18:30:10 +00:00
2021-08-01 14:12:34 +00:00
return read_ok;
2019-05-28 18:30:10 +00:00
}
2021-08-01 14:12:34 +00:00
void TCPHandler::readData()
2012-05-21 06:49:05 +00:00
{
2021-08-01 14:12:34 +00:00
sendLogs();
2021-08-01 14:12:34 +00:00
while (readDataNext())
;
2019-05-28 18:30:10 +00:00
}
2021-08-01 14:12:34 +00:00
void TCPHandler::skipData()
2019-05-28 18:30:10 +00:00
{
2021-08-01 14:12:34 +00:00
state.skipping_data = true;
SCOPE_EXIT({ state.skipping_data = false; });
2021-08-01 14:12:34 +00:00
while (readDataNext())
;
}
2021-08-01 14:12:34 +00:00
void TCPHandler::processInsertQuery()
{
2021-09-15 19:35:48 +00:00
size_t num_threads = state.io.pipeline.getNumThreads();
2021-09-10 14:52:24 +00:00
auto send_table_columns = [&]()
{
2021-09-10 14:52:24 +00:00
/// Send ColumnsDescription for insertion table
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_COLUMN_DEFAULTS_METADATA)
{
2021-09-10 14:52:24 +00:00
const auto & table_id = query_context->getInsertionTable();
if (query_context->getSettingsRef().input_format_defaults_for_omitted_fields)
{
2021-09-10 14:52:24 +00:00
if (!table_id.empty())
{
auto storage_ptr = DatabaseCatalog::instance().getTable(table_id, query_context);
sendTableColumns(storage_ptr->getInMemoryMetadataPtr()->getColumns());
}
}
}
2021-09-10 14:52:24 +00:00
};
if (num_threads > 1)
{
2021-09-15 19:35:48 +00:00
PushingAsyncPipelineExecutor executor(state.io.pipeline);
2021-09-10 14:52:24 +00:00
/** Made above the rest of the lines, so that in case of `writePrefix` function throws an exception,
* client receive exception before sending data.
*/
executor.start();
send_table_columns();
/// Send block to the client - table structure.
sendData(executor.getHeader());
sendLogs();
while (readDataNext())
executor.push(std::move(state.block_for_insert));
executor.finish();
}
2021-09-10 14:52:24 +00:00
else
{
2021-09-15 19:35:48 +00:00
PushingPipelineExecutor executor(state.io.pipeline);
2021-09-10 14:52:24 +00:00
executor.start();
2021-09-10 14:52:24 +00:00
send_table_columns();
2021-09-10 14:52:24 +00:00
sendData(executor.getHeader());
2021-09-10 14:52:24 +00:00
sendLogs();
2021-09-10 14:52:24 +00:00
while (readDataNext())
executor.push(std::move(state.block_for_insert));
executor.finish();
}
2012-05-21 06:49:05 +00:00
}
void TCPHandler::processOrdinaryQueryWithProcessors()
2019-03-26 18:28:37 +00:00
{
auto & pipeline = state.io.pipeline;
2021-02-05 10:15:02 +00:00
if (query_context->getSettingsRef().allow_experimental_query_deduplication)
sendPartUUIDs();
2019-03-26 18:28:37 +00:00
/// Send header-block, to allow client to prepare output format for data to send.
{
2020-05-14 21:03:38 +00:00
const auto & header = pipeline.getHeader();
2019-03-26 18:28:37 +00:00
if (header)
sendData(header);
}
{
2020-05-20 19:01:36 +00:00
PullingAsyncPipelineExecutor executor(pipeline);
2020-05-14 21:03:38 +00:00
CurrentMetrics::Increment query_thread_metric_increment{CurrentMetrics::QueryThread};
2020-05-14 21:03:38 +00:00
Block block;
2021-08-01 14:12:34 +00:00
while (executor.pull(block, interactive_delay / 1000))
2019-03-26 18:28:37 +00:00
{
2021-04-10 02:21:18 +00:00
std::lock_guard lock(task_callback_mutex);
2020-01-23 10:04:18 +00:00
if (isQueryCancelled())
2019-03-26 18:28:37 +00:00
{
2020-01-23 10:04:18 +00:00
/// A packet was received requesting to stop execution of the request.
2020-05-14 21:03:38 +00:00
executor.cancel();
2020-01-23 10:04:18 +00:00
break;
}
2019-03-26 18:28:37 +00:00
2021-08-01 14:12:34 +00:00
if (after_send_progress.elapsed() / 1000 >= interactive_delay)
2020-01-23 10:04:18 +00:00
{
/// Some time passed and there is a progress.
after_send_progress.restart();
sendProgress();
sendProfileEvents();
2020-01-23 10:04:18 +00:00
}
2019-03-26 18:28:37 +00:00
2020-01-23 10:04:18 +00:00
sendLogs();
2019-04-05 10:52:07 +00:00
2020-05-14 21:03:38 +00:00
if (block)
2020-01-23 10:04:18 +00:00
{
if (!state.io.null_format)
sendData(block);
2019-03-26 18:28:37 +00:00
}
2020-01-23 10:04:18 +00:00
}
2019-03-26 18:28:37 +00:00
2020-01-23 10:04:18 +00:00
/** If data has run out, we will send the profiling data and total values to
* the last zero block to be able to use
* this information in the suffix output of stream.
* If the request was interrupted, then `sendTotals` and other methods could not be called,
* because we have not read all the data yet,
* and there could be ongoing calculations in other threads at the same time.
*/
if (!isQueryCancelled())
{
2020-05-14 21:03:38 +00:00
sendTotals(executor.getTotalsBlock());
sendExtremes(executor.getExtremesBlock());
sendProfileInfo(executor.getProfileInfo());
2020-01-23 10:04:18 +00:00
sendProgress();
sendLogs();
sendProfileEvents();
2019-05-14 12:53:20 +00:00
}
2020-01-23 10:04:18 +00:00
if (state.is_connection_closed)
return;
2020-01-23 10:04:18 +00:00
sendData({});
2019-03-26 18:28:37 +00:00
}
2020-03-03 19:01:20 +00:00
sendProgress();
2019-03-26 18:28:37 +00:00
}
2012-07-21 07:02:55 +00:00
void TCPHandler::processTablesStatusRequest()
{
TablesStatusRequest request;
2020-09-17 12:15:05 +00:00
request.read(*in, client_tcp_protocol_version);
2021-08-01 14:12:34 +00:00
ContextPtr context_to_resolve_table_names = session->sessionContext() ? session->sessionContext() : server.context();
TablesStatusResponse response;
for (const QualifiedTableName & table_name: request.tables)
{
2021-08-01 14:12:34 +00:00
auto resolved_id = context_to_resolve_table_names->tryResolveStorageID({table_name.database, table_name.table});
StoragePtr table = DatabaseCatalog::instance().tryGetTable(resolved_id, context_to_resolve_table_names);
if (!table)
continue;
TableStatus status;
if (auto * replicated_table = dynamic_cast<StorageReplicatedMergeTree *>(table.get()))
{
status.is_replicated = true;
status.absolute_delay = replicated_table->getAbsoluteDelay();
}
else
status.is_replicated = false; //-V1048
response.table_states_by_id.emplace(table_name, std::move(status));
}
2021-02-17 17:34:52 +00:00
writeVarUInt(Protocol::Server::TablesStatusResponse, *out);
2021-01-19 19:21:06 +00:00
/// For testing hedged requests
2021-08-01 14:12:34 +00:00
if (sleep_in_send_tables_status.totalMilliseconds())
2021-01-19 19:21:06 +00:00
{
2021-02-17 17:34:52 +00:00
out->next();
2021-08-01 14:12:34 +00:00
std::chrono::milliseconds ms(sleep_in_send_tables_status.totalMilliseconds());
2021-03-22 21:21:52 +00:00
std::this_thread::sleep_for(ms);
2021-01-19 19:21:06 +00:00
}
2020-09-17 12:15:05 +00:00
response.write(*out, client_tcp_protocol_version);
}
2019-09-03 13:55:26 +00:00
void TCPHandler::receiveUnexpectedTablesStatusRequest()
{
TablesStatusRequest skip_request;
2020-09-17 12:15:05 +00:00
skip_request.read(*in, client_tcp_protocol_version);
throw NetException("Unexpected packet TablesStatusRequest received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
}
void TCPHandler::sendPartUUIDs()
{
auto uuids = query_context->getPartUUIDs()->get();
if (!uuids.empty())
{
for (const auto & uuid : uuids)
LOG_TRACE(log, "Sending UUID: {}", toString(uuid));
writeVarUInt(Protocol::Server::PartUUIDs, *out);
writeVectorBinary(uuids, *out);
out->next();
}
}
2021-03-22 17:12:31 +00:00
2021-04-10 02:21:18 +00:00
void TCPHandler::sendReadTaskRequestAssumeLocked()
2021-03-22 17:12:31 +00:00
{
writeVarUInt(Protocol::Server::ReadTaskRequest, *out);
2021-03-22 17:12:31 +00:00
out->next();
}
2019-03-26 18:28:37 +00:00
void TCPHandler::sendProfileInfo(const BlockStreamProfileInfo & info)
2013-05-22 14:57:43 +00:00
{
writeVarUInt(Protocol::Server::ProfileInfo, *out);
2019-03-26 18:28:37 +00:00
info.write(*out);
out->next();
2013-05-22 14:57:43 +00:00
}
2019-03-26 18:28:37 +00:00
void TCPHandler::sendTotals(const Block & totals)
{
if (totals)
{
initBlockOutput(totals);
writeVarUInt(Protocol::Server::Totals, *out);
writeStringBinary("", *out);
state.block_out->write(totals);
state.maybe_compressed_out->next();
out->next();
}
}
2019-03-26 18:28:37 +00:00
void TCPHandler::sendExtremes(const Block & extremes)
{
if (extremes)
{
initBlockOutput(extremes);
writeVarUInt(Protocol::Server::Extremes, *out);
writeStringBinary("", *out);
state.block_out->write(extremes);
state.maybe_compressed_out->next();
out->next();
}
}
2021-08-30 11:04:59 +00:00
namespace
{
using namespace ProfileEvents;
enum ProfileEventTypes : int8_t
{
INCREMENT = 1,
GAUGE = 2,
};
constexpr size_t NAME_COLUMN_INDEX = 4;
constexpr size_t VALUE_COLUMN_INDEX = 5;
/*
* Add records about provided non-zero ProfileEvents::Counters.
*/
void dumpProfileEvents(
ProfileEvents::Counters const & snapshot,
MutableColumns & columns,
String const & host_name,
time_t current_time,
UInt64 thread_id)
{
size_t rows = 0;
auto & name_column = columns[NAME_COLUMN_INDEX];
auto & value_column = columns[VALUE_COLUMN_INDEX];
for (ProfileEvents::Event event = 0; event < ProfileEvents::Counters::num_counters; ++event)
{
UInt64 value = snapshot[event].load(std::memory_order_relaxed);
if (value == 0)
continue;
const char * desc = ProfileEvents::getName(event);
name_column->insertData(desc, strlen(desc));
value_column->insert(value);
rows++;
}
// Fill the rest of the columns with data
for (size_t row = 0; row < rows; ++row)
{
size_t i = 0;
columns[i++]->insertData(host_name.data(), host_name.size());
columns[i++]->insert(UInt64(current_time));
columns[i++]->insert(UInt64{thread_id});
columns[i++]->insert(ProfileEventTypes::INCREMENT);
}
}
void dumpMemoryTracker(
MemoryTracker * memoryTracker,
MutableColumns & columns,
String const & host_name,
UInt64 thread_id)
{
auto metric = memoryTracker->getMetric();
if (metric == CurrentMetrics::end())
return;
time_t current_time = time(nullptr);
2021-08-30 11:04:59 +00:00
size_t i = 0;
columns[i++]->insertData(host_name.data(), host_name.size());
columns[i++]->insert(UInt64(current_time));
columns[i++]->insert(UInt64{thread_id});
columns[i++]->insert(ProfileEventTypes::GAUGE);
auto const * metric_name = CurrentMetrics::getName(metric);
columns[i++]->insertData(metric_name, strlen(metric_name));
auto metric_value = CurrentMetrics::get(metric);
columns[i++]->insert(metric_value);
}
}
void TCPHandler::sendProfileEvents()
{
auto profile_event_type = std::make_shared<DataTypeEnum8>(
DataTypeEnum8::Values
{
{ "increment", static_cast<Int8>(INCREMENT)},
{ "gauge", static_cast<Int8>(GAUGE)},
});
NamesAndTypesList column_names_and_types = {
{ "host_name", std::make_shared<DataTypeString>() },
{ "current_time", std::make_shared<DataTypeDateTime>() },
{ "thread_id", std::make_shared<DataTypeUInt64>() },
{ "type", profile_event_type },
{ "name", std::make_shared<DataTypeString>() },
{ "value", std::make_shared<DataTypeUInt64>() },
};
ColumnsWithTypeAndName temp_columns;
for (auto const & name_and_type : column_names_and_types)
temp_columns.emplace_back(name_and_type.type, name_and_type.name);
Block block(std::move(temp_columns));
MutableColumns columns = block.mutateColumns();
auto thread_group = CurrentThread::getGroup();
for (auto * thread : thread_group->threads)
{
auto const counters_snapshot = thread->performance_counters.getPartiallyAtomicSnapshot();
auto current_time = time(nullptr);
auto * memory_tracker = &thread->memory_tracker;
auto const thread_id = CurrentThread::get().thread_id;
2021-08-30 11:04:59 +00:00
dumpProfileEvents(counters_snapshot, columns, server_display_name, current_time, thread_id);
dumpMemoryTracker(memory_tracker, columns, server_display_name, thread_id);
}
2021-08-30 11:04:59 +00:00
block.setColumns(std::move(columns));
initProfileEventsBlockOutput(block);
writeVarUInt(Protocol::Server::ProfileEvents, *out);
writeStringBinary("", *out);
state.logs_block_out->write(block);
out->next();
}
2020-12-02 21:05:51 +00:00
bool TCPHandler::receiveProxyHeader()
{
if (in->eof())
{
LOG_WARNING(log, "Client has not sent any data.");
return false;
}
String forwarded_address;
/// Only PROXYv1 is supported.
/// Validation of protocol is not fully performed.
LimitReadBuffer limit_in(*in, 107, true); /// Maximum length from the specs.
assertString("PROXY ", limit_in);
if (limit_in.eof())
{
LOG_WARNING(log, "Incomplete PROXY header is received.");
return false;
}
/// TCP4 / TCP6 / UNKNOWN
if ('T' == *limit_in.position())
{
assertString("TCP", limit_in);
if (limit_in.eof())
{
LOG_WARNING(log, "Incomplete PROXY header is received.");
return false;
}
if ('4' != *limit_in.position() && '6' != *limit_in.position())
{
LOG_WARNING(log, "Unexpected protocol in PROXY header is received.");
return false;
}
++limit_in.position();
assertChar(' ', limit_in);
/// Read the first field and ignore other.
readStringUntilWhitespace(forwarded_address, limit_in);
/// Skip until \r\n
while (!limit_in.eof() && *limit_in.position() != '\r')
++limit_in.position();
assertString("\r\n", limit_in);
}
else if (checkString("UNKNOWN", limit_in))
{
/// This is just a health check, there is no subsequent data in this connection.
while (!limit_in.eof() && *limit_in.position() != '\r')
++limit_in.position();
assertString("\r\n", limit_in);
return false;
}
else
{
LOG_WARNING(log, "Unexpected protocol in PROXY header is received.");
return false;
}
LOG_TRACE(log, "Forwarded client address from PROXY header: {}", forwarded_address);
session->getClientInfo().forwarded_for = forwarded_address;
2020-12-02 21:05:51 +00:00
return true;
}
namespace
{
std::string formatHTTPErrorResponseWhenUserIsConnectedToWrongPort(const Poco::Util::AbstractConfiguration& config)
{
std::string result = fmt::format(
"HTTP/1.0 400 Bad Request\r\n\r\n"
"Port {} is for clickhouse-client program\r\n",
config.getString("tcp_port"));
if (config.has("http_port"))
{
result += fmt::format(
"You must use port {} for HTTP.\r\n",
config.getString("http_port"));
}
return result;
}
}
2012-05-21 06:49:05 +00:00
void TCPHandler::receiveHello()
2012-05-16 18:03:00 +00:00
{
/// Receive `hello` packet.
UInt64 packet_type = 0;
String user;
String password;
readVarUInt(packet_type, *in);
if (packet_type != Protocol::Client::Hello)
{
/** If you accidentally accessed the HTTP protocol for a port destined for an internal TCP protocol,
* Then instead of the packet type, there will be G (GET) or P (POST), in most cases.
*/
if (packet_type == 'G' || packet_type == 'P')
{
writeString(formatHTTPErrorResponseWhenUserIsConnectedToWrongPort(server.config()), *out);
throw Exception("Client has connected to wrong port", ErrorCodes::CLIENT_HAS_CONNECTED_TO_WRONG_PORT);
}
else
throw NetException("Unexpected packet from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
}
readStringBinary(client_name, *in);
readVarUInt(client_version_major, *in);
readVarUInt(client_version_minor, *in);
// NOTE For backward compatibility of the protocol, client cannot send its version_patch.
2020-09-17 12:15:05 +00:00
readVarUInt(client_tcp_protocol_version, *in);
readStringBinary(default_database, *in);
readStringBinary(user, *in);
readStringBinary(password, *in);
if (user.empty())
throw NetException("Unexpected packet from client (no user in Hello package)", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
2020-05-23 22:24:01 +00:00
LOG_DEBUG(log, "Connected {} version {}.{}.{}, revision: {}{}{}.",
2020-05-23 22:21:29 +00:00
client_name,
client_version_major, client_version_minor, client_version_patch,
2020-09-17 12:15:05 +00:00
client_tcp_protocol_version,
2020-05-23 22:21:29 +00:00
(!default_database.empty() ? ", database: " + default_database : ""),
(!user.empty() ? ", user: " + user : "")
);
2021-08-01 14:12:34 +00:00
auto & client_info = session->getClientInfo();
client_info.client_name = client_name;
client_info.client_version_major = client_version_major;
client_info.client_version_minor = client_version_minor;
client_info.client_version_patch = client_version_patch;
client_info.client_tcp_protocol_version = client_tcp_protocol_version;
is_interserver_mode = (user == USER_INTERSERVER_MARKER);
if (is_interserver_mode)
{
client_info.interface = ClientInfo::Interface::TCP_INTERSERVER;
receiveClusterNameAndSalt();
2021-08-01 14:12:34 +00:00
return;
}
2021-08-01 14:12:34 +00:00
session->authenticate(user, password, socket().peerAddress());
2012-05-16 18:03:00 +00:00
}
void TCPHandler::receiveUnexpectedHello()
{
UInt64 skip_uint_64;
String skip_string;
readStringBinary(skip_string, *in);
readVarUInt(skip_uint_64, *in);
readVarUInt(skip_uint_64, *in);
2020-04-15 01:58:10 +00:00
readVarUInt(skip_uint_64, *in);
readStringBinary(skip_string, *in);
readStringBinary(skip_string, *in);
readStringBinary(skip_string, *in);
throw NetException("Unexpected packet Hello received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
}
2012-05-21 06:49:05 +00:00
void TCPHandler::sendHello()
2012-03-11 08:52:56 +00:00
{
writeVarUInt(Protocol::Server::Hello, *out);
writeStringBinary(DBMS_NAME, *out);
writeVarUInt(DBMS_VERSION_MAJOR, *out);
writeVarUInt(DBMS_VERSION_MINOR, *out);
2020-09-17 12:15:05 +00:00
writeVarUInt(DBMS_TCP_PROTOCOL_VERSION, *out);
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_SERVER_TIMEZONE)
writeStringBinary(DateLUT::instance().getTimeZone(), *out);
2020-09-17 12:15:05 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_SERVER_DISPLAY_NAME)
writeStringBinary(server_display_name, *out);
2020-09-17 12:15:05 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_VERSION_PATCH)
writeVarUInt(DBMS_VERSION_PATCH, *out);
out->next();
2012-03-11 08:52:56 +00:00
}
2012-05-17 19:15:53 +00:00
2012-05-21 06:49:05 +00:00
bool TCPHandler::receivePacket()
2012-03-11 08:52:56 +00:00
{
UInt64 packet_type = 0;
2021-04-06 19:18:45 +00:00
readVarUInt(packet_type, *in);
2021-03-19 21:49:18 +00:00
switch (packet_type)
{
case Protocol::Client::IgnoredPartUUIDs:
/// Part uuids packet if any comes before query.
2021-08-01 14:12:34 +00:00
if (!state.empty() || state.part_uuids_to_ignore)
receiveUnexpectedIgnoredPartUUIDs();
receiveIgnoredPartUUIDs();
return true;
2021-08-01 14:12:34 +00:00
case Protocol::Client::Query:
if (!state.empty())
receiveUnexpectedQuery();
receiveQuery();
return true;
case Protocol::Client::Data:
2019-10-19 20:36:35 +00:00
case Protocol::Client::Scalar:
2021-08-01 14:12:34 +00:00
if (state.skipping_data)
return receiveUnexpectedData(false);
if (state.empty())
2021-08-01 14:12:34 +00:00
receiveUnexpectedData(true);
2019-10-19 20:36:35 +00:00
return receiveData(packet_type == Protocol::Client::Scalar);
case Protocol::Client::Ping:
writeVarUInt(Protocol::Server::Pong, *out);
out->next();
return false;
case Protocol::Client::Cancel:
2021-07-14 13:17:30 +00:00
{
/// For testing connection collector.
2021-08-01 14:12:34 +00:00
if (sleep_in_receive_cancel.totalMilliseconds())
2021-07-14 13:17:30 +00:00
{
2021-08-01 14:12:34 +00:00
std::chrono::milliseconds ms(sleep_in_receive_cancel.totalMilliseconds());
2021-07-14 13:17:30 +00:00
std::this_thread::sleep_for(ms);
}
return false;
2021-07-14 13:17:30 +00:00
}
case Protocol::Client::Hello:
receiveUnexpectedHello();
case Protocol::Client::TablesStatusRequest:
if (!state.empty())
2019-09-03 13:55:26 +00:00
receiveUnexpectedTablesStatusRequest();
processTablesStatusRequest();
out->next();
return false;
default:
throw Exception("Unknown packet " + toString(packet_type) + " from client", ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT);
}
2012-03-09 15:46:52 +00:00
}
2021-08-01 14:12:34 +00:00
void TCPHandler::receiveIgnoredPartUUIDs()
{
2021-08-01 14:12:34 +00:00
readVectorBinary(state.part_uuids_to_ignore.emplace(), *in);
}
2021-08-01 14:12:34 +00:00
void TCPHandler::receiveUnexpectedIgnoredPartUUIDs()
{
std::vector<UUID> skip_part_uuids;
readVectorBinary(skip_part_uuids, *in);
throw NetException("Unexpected packet IgnoredPartUUIDs received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
}
2021-04-13 10:59:02 +00:00
String TCPHandler::receiveReadTaskResponseAssumeLocked()
{
UInt64 packet_type = 0;
readVarUInt(packet_type, *in);
if (packet_type != Protocol::Client::ReadTaskResponse)
2021-04-10 02:21:18 +00:00
{
if (packet_type == Protocol::Client::Cancel)
{
state.is_cancelled = true;
2021-07-14 13:17:30 +00:00
/// For testing connection collector.
2021-08-01 14:12:34 +00:00
if (sleep_in_receive_cancel.totalMilliseconds())
2021-07-14 13:17:30 +00:00
{
2021-08-01 14:12:34 +00:00
std::chrono::milliseconds ms(sleep_in_receive_cancel.totalMilliseconds());
2021-07-14 13:17:30 +00:00
std::this_thread::sleep_for(ms);
}
2021-04-10 02:21:18 +00:00
return {};
}
else
{
throw Exception(fmt::format("Received {} packet after requesting read task",
Protocol::Client::toString(packet_type)), ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
}
}
UInt64 version;
readVarUInt(version, *in);
if (version != DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION)
2021-04-12 17:48:16 +00:00
throw Exception("Protocol version for distributed processing mismatched", ErrorCodes::UNKNOWN_PROTOCOL);
String response;
readStringBinary(response, *in);
return response;
}
void TCPHandler::receiveClusterNameAndSalt()
{
2021-04-06 19:18:45 +00:00
readStringBinary(cluster, *in);
readStringBinary(salt, *in, 32);
try
{
if (salt.empty())
throw NetException("Empty salt is not allowed", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
2021-08-01 14:12:34 +00:00
cluster_secret = server.context()->getCluster(cluster)->getSecret();
}
catch (const Exception & e)
{
try
{
/// We try to send error information to the client.
2021-08-01 14:12:34 +00:00
sendException(e, send_exception_with_stack_trace);
}
catch (...) {}
throw;
}
}
2012-05-17 19:15:53 +00:00
2012-05-21 06:49:05 +00:00
void TCPHandler::receiveQuery()
2012-03-11 08:52:56 +00:00
{
UInt64 stage = 0;
UInt64 compression = 0;
state.is_empty = false;
readStringBinary(state.query_id, *in);
Do not allow to reuse previous credentials in case of inter-server secret Before this patch INSERT via Buffer/Kafka may re-use previously set user for that connection, while this is not correct, it should reset the user, and use global context. Note, before [1] there was a fallback to default user, but that code had been removed, and now it got back. [1]: 0159c74f217ec764060c480819e3ccc9d5a99a63 ("Secure inter-cluster query execution (with initial_user as current query user) [v3]") Also note, that context for Buffer table (and others) cannot be changed, since they don't have any user only profile. I've tested this patch manually using the following: create table dist (key Int) engine=Distributed(test_cluster_two_shards_secure, default, data, key); create table buffer (key Int) engine=Buffer(default, dist, 1, 0, 0, 0, 0, 0, 0); create table data (key Int) engine=Memory(); # to start the connection with readonly user $ clickhouse-client --user readonly -q 'select * from dist' $ clickhouse-client -q 'insert into buffer values (1)' # before this patch this produces errors like: # 2021.09.27 23:46:48.384920 [ 19474 ] {} <Error> default.dist.DirectoryMonitor: Code: 164. DB::Exception: Received from 127.0.0.2:9000. DB::Exception: readonly: Cannot execute query in readonly mode. Stack trace: v2: reset the authentication instead of using default user (as suggested by @vitlibar) v3: reset Session::user and introduce ClientInfo::resetAuthentication (as suggested by @vitlibar) v4: reset the session every time in interserver mode (suggested by @vitlibar)
2021-09-30 07:15:38 +00:00
/// In interserer mode,
/// initial_user can be empty in case of Distributed INSERT via Buffer/Kafka,
/// (i.e. when the INSERT is done with the global context w/o user),
/// so it is better to reset session to avoid using old user.
if (is_interserver_mode)
{
ClientInfo original_session_client_info = session->getClientInfo();
session = std::make_unique<Session>(server.context(), ClientInfo::Interface::TCP_INTERSERVER);
session->getClientInfo() = original_session_client_info;
}
2021-08-01 14:12:34 +00:00
/// Read client info.
ClientInfo client_info = session->getClientInfo();
2020-09-17 12:15:05 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_CLIENT_INFO)
client_info.read(*in, client_tcp_protocol_version);
/// Per query settings are also passed via TCP.
/// We need to check them before applying due to they can violate the settings constraints.
auto settings_format = (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS)
? SettingsWriteFormat::STRINGS_WITH_FLAGS
: SettingsWriteFormat::BINARY;
Settings passed_settings;
passed_settings.read(*in, settings_format);
/// Interserver secret.
std::string received_hash;
2020-09-17 14:55:41 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET)
{
readStringBinary(received_hash, *in, 32);
}
readVarUInt(stage, *in);
state.stage = QueryProcessingStage::Enum(stage);
readVarUInt(compression, *in);
state.compression = static_cast<Protocol::Compression>(compression);
2021-08-01 14:12:34 +00:00
last_block_in.compression = state.compression;
readStringBinary(state.query, *in);
2021-08-01 14:12:34 +00:00
if (is_interserver_mode)
{
#if USE_SSL
std::string data(salt);
data += cluster_secret;
data += state.query;
data += state.query_id;
data += client_info.initial_user;
if (received_hash.size() != 32)
throw NetException("Unexpected hash received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
std::string calculated_hash = encodeSHA256(data);
if (calculated_hash != received_hash)
throw NetException("Hash mismatch", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
/// TODO: change error code?
Do not allow to reuse previous credentials in case of inter-server secret Before this patch INSERT via Buffer/Kafka may re-use previously set user for that connection, while this is not correct, it should reset the user, and use global context. Note, before [1] there was a fallback to default user, but that code had been removed, and now it got back. [1]: 0159c74f217ec764060c480819e3ccc9d5a99a63 ("Secure inter-cluster query execution (with initial_user as current query user) [v3]") Also note, that context for Buffer table (and others) cannot be changed, since they don't have any user only profile. I've tested this patch manually using the following: create table dist (key Int) engine=Distributed(test_cluster_two_shards_secure, default, data, key); create table buffer (key Int) engine=Buffer(default, dist, 1, 0, 0, 0, 0, 0, 0); create table data (key Int) engine=Memory(); # to start the connection with readonly user $ clickhouse-client --user readonly -q 'select * from dist' $ clickhouse-client -q 'insert into buffer values (1)' # before this patch this produces errors like: # 2021.09.27 23:46:48.384920 [ 19474 ] {} <Error> default.dist.DirectoryMonitor: Code: 164. DB::Exception: Received from 127.0.0.2:9000. DB::Exception: readonly: Cannot execute query in readonly mode. Stack trace: v2: reset the authentication instead of using default user (as suggested by @vitlibar) v3: reset Session::user and introduce ClientInfo::resetAuthentication (as suggested by @vitlibar) v4: reset the session every time in interserver mode (suggested by @vitlibar)
2021-09-30 07:15:38 +00:00
if (client_info.initial_user.empty())
{
LOG_DEBUG(log, "User (no user, interserver mode)");
}
else
{
Do not allow to reuse previous credentials in case of inter-server secret Before this patch INSERT via Buffer/Kafka may re-use previously set user for that connection, while this is not correct, it should reset the user, and use global context. Note, before [1] there was a fallback to default user, but that code had been removed, and now it got back. [1]: 0159c74f217ec764060c480819e3ccc9d5a99a63 ("Secure inter-cluster query execution (with initial_user as current query user) [v3]") Also note, that context for Buffer table (and others) cannot be changed, since they don't have any user only profile. I've tested this patch manually using the following: create table dist (key Int) engine=Distributed(test_cluster_two_shards_secure, default, data, key); create table buffer (key Int) engine=Buffer(default, dist, 1, 0, 0, 0, 0, 0, 0); create table data (key Int) engine=Memory(); # to start the connection with readonly user $ clickhouse-client --user readonly -q 'select * from dist' $ clickhouse-client -q 'insert into buffer values (1)' # before this patch this produces errors like: # 2021.09.27 23:46:48.384920 [ 19474 ] {} <Error> default.dist.DirectoryMonitor: Code: 164. DB::Exception: Received from 127.0.0.2:9000. DB::Exception: readonly: Cannot execute query in readonly mode. Stack trace: v2: reset the authentication instead of using default user (as suggested by @vitlibar) v3: reset Session::user and introduce ClientInfo::resetAuthentication (as suggested by @vitlibar) v4: reset the session every time in interserver mode (suggested by @vitlibar)
2021-09-30 07:15:38 +00:00
LOG_DEBUG(log, "User (initial, interserver mode): {}", client_info.initial_user);
2021-08-01 14:12:34 +00:00
session->authenticate(AlwaysAllowCredentials{client_info.initial_user}, client_info.initial_address);
}
#else
throw Exception(
"Inter-server secret support is disabled, because ClickHouse was built without SSL library",
ErrorCodes::SUPPORT_IS_DISABLED);
#endif
}
2021-08-01 14:12:34 +00:00
query_context = session->makeQueryContext(std::move(client_info));
/// Sets the default database if it wasn't set earlier for the session context.
if (!default_database.empty() && !session->sessionContext())
query_context->setCurrentDatabase(default_database);
if (state.part_uuids_to_ignore)
query_context->getIgnoredPartUUIDs()->add(*state.part_uuids_to_ignore);
query_context->setProgressCallback([this] (const Progress & value) { return this->updateProgress(value); });
///
/// Settings
///
auto settings_changes = passed_settings.changes();
2021-08-01 14:12:34 +00:00
auto query_kind = query_context->getClientInfo().query_kind;
if (query_kind == ClientInfo::QueryKind::INITIAL_QUERY)
{
/// Throw an exception if the passed settings violate the constraints.
query_context->checkSettingsConstraints(settings_changes);
}
else
{
/// Quietly clamp to the constraints if it's not an initial query.
query_context->clampToSettingsConstraints(settings_changes);
}
query_context->applySettingsChanges(settings_changes);
2020-09-08 13:19:27 +00:00
2021-08-01 14:12:34 +00:00
/// Use the received query id, or generate a random default. It is convenient
/// to also generate the default OpenTelemetry trace id at the same time, and
/// set the trace parent.
/// Notes:
/// 1) ClientInfo might contain upstream trace id, so we decide whether to use
/// the default ids after we have received the ClientInfo.
/// 2) There is the opentelemetry_start_trace_probability setting that
/// controls when we start a new trace. It can be changed via Native protocol,
/// so we have to apply the changes first.
query_context->setCurrentQueryId(state.query_id);
2021-02-20 08:45:25 +00:00
/// Disable function name normalization when it's a secondary query, because queries are either
/// already normalized on initiator node, or not normalized and should remain unnormalized for
/// compatibility.
2021-08-01 14:12:34 +00:00
if (query_kind == ClientInfo::QueryKind::SECONDARY_QUERY)
2020-10-19 21:26:10 +00:00
{
query_context->setSetting("normalize_function_names", false);
2020-10-19 21:26:10 +00:00
}
2012-03-11 08:52:56 +00:00
}
void TCPHandler::receiveUnexpectedQuery()
{
UInt64 skip_uint_64;
String skip_string;
readStringBinary(skip_string, *in);
ClientInfo skip_client_info;
2020-09-17 12:15:05 +00:00
if (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_CLIENT_INFO)
skip_client_info.read(*in, client_tcp_protocol_version);
Settings skip_settings;
2020-09-17 12:15:05 +00:00
auto settings_format = (client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_SETTINGS_SERIALIZED_AS_STRINGS) ? SettingsWriteFormat::STRINGS_WITH_FLAGS
: SettingsWriteFormat::BINARY;
skip_settings.read(*in, settings_format);
std::string skip_hash;
2020-09-17 14:55:41 +00:00
bool interserver_secret = client_tcp_protocol_version >= DBMS_MIN_REVISION_WITH_INTERSERVER_SECRET;
if (interserver_secret)
readStringBinary(skip_hash, *in, 32);
readVarUInt(skip_uint_64, *in);
2021-08-01 14:12:34 +00:00
readVarUInt(skip_uint_64, *in);
2021-08-01 14:12:34 +00:00
last_block_in.compression = static_cast<Protocol::Compression>(skip_uint_64);
readStringBinary(skip_string, *in);
throw NetException("Unexpected packet Query received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
}
2012-05-21 06:49:05 +00:00
2019-10-19 20:36:35 +00:00
bool TCPHandler::receiveData(bool scalar)
{
initBlockInput();
/// The name of the temporary table for writing data, default to empty string
2020-02-12 18:14:12 +00:00
auto temporary_id = StorageID::createEmpty();
readStringBinary(temporary_id.table_name, *in);
/// Read one block from the network and write it down
Block block = state.block_in->read();
2021-08-01 14:12:34 +00:00
if (!block)
{
2021-08-01 14:12:34 +00:00
state.read_all_data = true;
return false;
}
2021-02-10 22:23:27 +00:00
2021-08-01 14:12:34 +00:00
if (scalar)
{
/// Scalar value
query_context->addScalar(temporary_id.table_name, block);
}
else if (!state.need_receive_data_for_insert && !state.need_receive_data_for_input)
{
/// Data for external tables
2021-02-10 22:23:27 +00:00
2021-08-01 14:12:34 +00:00
auto resolved = query_context->tryResolveStorageID(temporary_id, Context::ResolveExternal);
StoragePtr storage;
/// If such a table does not exist, create it.
if (resolved)
2021-02-10 22:23:27 +00:00
{
2021-08-01 14:12:34 +00:00
storage = DatabaseCatalog::instance().getTable(resolved, query_context);
2021-02-10 22:23:27 +00:00
}
else
{
2021-08-01 14:12:34 +00:00
NamesAndTypesList columns = block.getNamesAndTypesList();
auto temporary_table = TemporaryTableHolder(query_context, ColumnsDescription{columns}, {});
storage = temporary_table.getTable();
query_context->addExternalTable(temporary_id.table_name, std::move(temporary_table));
}
2021-08-01 14:12:34 +00:00
auto metadata_snapshot = storage->getInMemoryMetadataPtr();
/// The data will be written directly to the table.
2021-09-16 17:40:42 +00:00
QueryPipeline temporary_table_out(storage->write(ASTPtr(), metadata_snapshot, query_context));
PushingPipelineExecutor executor(temporary_table_out);
executor.start();
executor.push(block);
executor.finish();
2021-08-01 14:12:34 +00:00
}
else if (state.need_receive_data_for_input)
{
/// 'input' table function.
state.block_for_input = block;
}
else
2021-08-01 14:12:34 +00:00
{
/// INSERT query.
state.block_for_insert = block;
2021-08-01 14:12:34 +00:00
}
return true;
}
2021-08-01 14:12:34 +00:00
bool TCPHandler::receiveUnexpectedData(bool throw_exception)
{
String skip_external_table_name;
readStringBinary(skip_external_table_name, *in);
std::shared_ptr<ReadBuffer> maybe_compressed_in;
if (last_block_in.compression == Protocol::Compression::Enable)
maybe_compressed_in = std::make_shared<CompressedReadBuffer>(*in, /* allow_different_codecs */ true);
else
maybe_compressed_in = in;
2021-10-08 17:21:19 +00:00
auto skip_block_in = std::make_shared<NativeReader>(*maybe_compressed_in, client_tcp_protocol_version);
2021-08-01 14:12:34 +00:00
bool read_ok = skip_block_in->read();
2021-08-01 14:12:34 +00:00
if (!read_ok)
state.read_all_data = true;
2021-08-01 14:12:34 +00:00
if (throw_exception)
throw NetException("Unexpected packet Data received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
return read_ok;
}
void TCPHandler::initBlockInput()
2012-03-11 08:52:56 +00:00
{
if (!state.block_in)
{
/// 'allow_different_codecs' is set to true, because some parts of compressed data can be precompressed in advance
/// with another codec that the rest of the data. Example: data sent by Distributed tables.
if (state.compression == Protocol::Compression::Enable)
state.maybe_compressed_in = std::make_shared<CompressedReadBuffer>(*in, /* allow_different_codecs */ true);
else
state.maybe_compressed_in = in;
Block header;
2021-09-16 17:40:42 +00:00
if (state.io.pipeline.pushing())
header = state.io.pipeline.getHeader();
2019-05-28 18:30:10 +00:00
else if (state.need_receive_data_for_input)
header = state.input_header;
2021-10-08 17:21:19 +00:00
state.block_in = std::make_unique<NativeReader>(
*state.maybe_compressed_in,
header,
2020-09-17 12:15:05 +00:00
client_tcp_protocol_version);
}
}
void TCPHandler::initBlockOutput(const Block & block)
{
if (!state.block_out)
{
2021-08-01 14:12:34 +00:00
const Settings & query_settings = query_context->getSettingsRef();
if (!state.maybe_compressed_out)
{
2020-05-04 00:11:49 +00:00
std::string method = Poco::toUpper(query_settings.network_compression_method.toString());
2018-12-21 12:17:30 +00:00
std::optional<int> level;
if (method == "ZSTD")
2020-05-04 00:11:49 +00:00
level = query_settings.network_zstd_compression_level;
2018-12-21 12:17:30 +00:00
if (state.compression == Protocol::Compression::Enable)
{
CompressionCodecFactory::instance().validateCodec(method, level, !query_settings.allow_suspicious_codecs, query_settings.allow_experimental_codecs);
state.maybe_compressed_out = std::make_shared<CompressedWriteBuffer>(
2021-05-24 03:43:25 +00:00
*out, CompressionCodecFactory::instance().get(method, level));
}
else
state.maybe_compressed_out = out;
}
2021-10-08 17:21:19 +00:00
state.block_out = std::make_unique<NativeWriter>(
*state.maybe_compressed_out,
2020-09-17 12:15:05 +00:00
client_tcp_protocol_version,
block.cloneEmpty(),
2021-08-01 14:12:34 +00:00
!query_settings.low_cardinality_allow_in_native_format);
}
2012-03-19 12:57:56 +00:00
}
2012-03-11 08:52:56 +00:00
void TCPHandler::initLogsBlockOutput(const Block & block)
{
if (!state.logs_block_out)
{
/// Use uncompressed stream since log blocks usually contain only one row
2021-08-01 14:12:34 +00:00
const Settings & query_settings = query_context->getSettingsRef();
2021-10-08 17:21:19 +00:00
state.logs_block_out = std::make_unique<NativeWriter>(
*out,
2020-09-17 12:15:05 +00:00
client_tcp_protocol_version,
block.cloneEmpty(),
2021-08-01 14:12:34 +00:00
!query_settings.low_cardinality_allow_in_native_format);
}
}
2021-08-30 11:04:59 +00:00
void TCPHandler::initProfileEventsBlockOutput(const Block & block)
{
if (!state.profile_events_block_out)
{
const Settings & query_settings = query_context->getSettingsRef();
state.profile_events_block_out = std::make_unique<NativeWriter>(
*out,
client_tcp_protocol_version,
block.cloneEmpty(),
!query_settings.low_cardinality_allow_in_native_format);
}
}
2012-05-21 06:49:05 +00:00
bool TCPHandler::isQueryCancelled()
2012-05-09 08:16:09 +00:00
{
if (state.is_cancelled || state.sent_all_data)
return true;
2021-08-01 14:12:34 +00:00
if (after_check_cancelled.elapsed() / 1000 < interactive_delay)
return false;
after_check_cancelled.restart();
/// During request execution the only packet that can come from the client is stopping the query.
if (static_cast<ReadBufferFromPocoSocket &>(*in).poll(0))
{
if (in->eof())
{
LOG_INFO(log, "Client has dropped the connection, cancel the query.");
state.is_cancelled = true;
state.is_connection_closed = true;
return true;
}
UInt64 packet_type = 0;
readVarUInt(packet_type, *in);
switch (packet_type)
{
case Protocol::Client::Cancel:
if (state.empty())
throw NetException("Unexpected packet Cancel received from client", ErrorCodes::UNEXPECTED_PACKET_FROM_CLIENT);
LOG_INFO(log, "Query was cancelled.");
state.is_cancelled = true;
2021-07-14 13:17:30 +00:00
/// For testing connection collector.
{
2021-08-01 14:12:34 +00:00
if (sleep_in_receive_cancel.totalMilliseconds())
2021-07-14 13:17:30 +00:00
{
2021-08-01 14:12:34 +00:00
std::chrono::milliseconds ms(sleep_in_receive_cancel.totalMilliseconds());
2021-07-14 13:17:30 +00:00
std::this_thread::sleep_for(ms);
}
}
return true;
default:
throw NetException("Unknown packet from client", ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT);
}
}
return false;
2012-05-09 08:16:09 +00:00
}
2012-05-21 06:49:05 +00:00
void TCPHandler::sendData(const Block & block)
2012-03-19 12:57:56 +00:00
{
initBlockOutput(block);
2012-03-11 08:52:56 +00:00
auto prev_bytes_written_out = out->count();
auto prev_bytes_written_compressed_out = state.maybe_compressed_out->count();
2012-05-21 06:49:05 +00:00
try
2021-02-17 17:34:52 +00:00
{
2021-04-19 14:12:08 +00:00
/// For testing hedged requests
2021-04-19 17:55:18 +00:00
if (unknown_packet_in_send_data)
2021-04-19 14:12:08 +00:00
{
2021-04-19 17:55:18 +00:00
--unknown_packet_in_send_data;
if (unknown_packet_in_send_data == 0)
2021-04-19 14:12:08 +00:00
writeVarUInt(UInt64(-1), *out);
}
writeVarUInt(Protocol::Server::Data, *out);
/// Send external table name (empty name is the main table)
writeStringBinary("", *out);
/// For testing hedged requests
2021-08-01 14:12:34 +00:00
if (block.rows() > 0 && query_context->getSettingsRef().sleep_in_send_data_ms.totalMilliseconds())
{
out->next();
2021-08-01 14:12:34 +00:00
std::chrono::milliseconds ms(query_context->getSettingsRef().sleep_in_send_data_ms.totalMilliseconds());
std::this_thread::sleep_for(ms);
}
state.block_out->write(block);
state.maybe_compressed_out->next();
2021-02-17 17:34:52 +00:00
out->next();
}
catch (...)
{
/// In case of unsuccessful write, if the buffer with written data was not flushed,
/// we will rollback write to avoid breaking the protocol.
/// (otherwise the client will not be able to receive exception after unfinished data
/// as it will expect the continuation of the data).
/// It looks like hangs on client side or a message like "Data compressed with different methods".
2021-02-17 17:34:52 +00:00
if (state.compression == Protocol::Compression::Enable)
{
auto extra_bytes_written_compressed = state.maybe_compressed_out->count() - prev_bytes_written_compressed_out;
if (state.maybe_compressed_out->offset() >= extra_bytes_written_compressed)
state.maybe_compressed_out->position() -= extra_bytes_written_compressed;
}
auto extra_bytes_written_out = out->count() - prev_bytes_written_out;
if (out->offset() >= extra_bytes_written_out)
out->position() -= extra_bytes_written_out;
throw;
}
2012-03-19 12:57:56 +00:00
}
2012-05-21 06:49:05 +00:00
void TCPHandler::sendLogData(const Block & block)
{
initLogsBlockOutput(block);
writeVarUInt(Protocol::Server::Log, *out);
/// Send log tag (empty tag is the default tag)
writeStringBinary("", *out);
state.logs_block_out->write(block);
out->next();
}
void TCPHandler::sendTableColumns(const ColumnsDescription & columns)
{
writeVarUInt(Protocol::Server::TableColumns, *out);
/// Send external table name (empty name is the main table)
writeStringBinary("", *out);
writeStringBinary(columns.toString(), *out);
out->next();
}
2018-08-24 07:30:53 +00:00
void TCPHandler::sendException(const Exception & e, bool with_stack_trace)
2012-03-19 12:57:56 +00:00
{
writeVarUInt(Protocol::Server::Exception, *out);
2018-08-24 07:30:53 +00:00
writeException(e, *out, with_stack_trace);
out->next();
2012-03-19 12:57:56 +00:00
}
2012-05-21 06:49:05 +00:00
void TCPHandler::sendEndOfStream()
2012-05-08 11:19:00 +00:00
{
state.sent_all_data = true;
writeVarUInt(Protocol::Server::EndOfStream, *out);
out->next();
2012-05-08 11:19:00 +00:00
}
2012-05-21 06:49:05 +00:00
void TCPHandler::updateProgress(const Progress & value)
2012-03-19 12:57:56 +00:00
{
state.progress.incrementPiecewiseAtomically(value);
}
2012-05-09 15:15:45 +00:00
void TCPHandler::sendProgress()
{
writeVarUInt(Protocol::Server::Progress, *out);
auto increment = state.progress.fetchAndResetPiecewiseAtomically();
2020-09-17 12:15:05 +00:00
increment.write(*out, client_tcp_protocol_version);
out->next();
2012-03-11 08:52:56 +00:00
}
2012-03-09 15:46:52 +00:00
void TCPHandler::sendLogs()
{
if (!state.logs_queue)
return;
MutableColumns logs_columns;
MutableColumns curr_logs_columns;
size_t rows = 0;
for (; state.logs_queue->tryPop(curr_logs_columns); ++rows)
{
if (rows == 0)
{
logs_columns = std::move(curr_logs_columns);
}
else
{
for (size_t j = 0; j < logs_columns.size(); ++j)
logs_columns[j]->insertRangeFrom(*curr_logs_columns[j], 0, curr_logs_columns[j]->size());
}
}
if (rows > 0)
{
Block block = InternalTextLogsQueue::getSampleBlock();
block.setColumns(std::move(logs_columns));
sendLogData(block);
}
}
2012-03-09 15:46:52 +00:00
void TCPHandler::run()
{
try
{
runImpl();
2020-10-10 17:47:34 +00:00
LOG_DEBUG(log, "Done processing connection.");
}
catch (Poco::Exception & e)
{
/// Timeout - not an error.
if (!strcmp(e.what(), "Timeout"))
{
2020-05-23 22:24:01 +00:00
LOG_DEBUG(log, "Poco::Exception. Code: {}, e.code() = {}, e.displayText() = {}, e.what() = {}", ErrorCodes::POCO_EXCEPTION, e.code(), e.displayText(), e.what());
}
else
throw;
}
2012-03-09 15:46:52 +00:00
}
}