thevar1able/ClickHouse
1
0
Fork 0
mirror of https://github.com/ClickHouse/ClickHouse.git synced 2024-09-22 01:30:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into try-to-remove-dry-run

Browse Source
This commit is contained in:
Nikolai Kochetov 2019-08-19 09:09:22 +03:00
commit b8a5e107df
27 changed files with 641 additions and 286 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
dbms/cmake/version.cmake
View File

@ -1,11 +1,11 @@
# This strings autochanged from release_lib.sh:
set(VERSION_REVISION 54425)
set(VERSION_MAJOR 19)
set(VERSION_MINOR 13)
set(VERSION_MINOR 14)
set(VERSION_PATCH 1)
set(VERSION_GITHASH adfc36917222bdb03eba069f0cad0f4f5b8f1c94)
set(VERSION_DESCRIBE v19.13.1.1-prestable)
set(VERSION_STRING 19.13.1.1)
set(VERSION_DESCRIBE v19.14.1.1-prestable)
set(VERSION_STRING 19.14.1.1)
# end of autochange
set(VERSION_EXTRA "" CACHE STRING "")

140
dbms/programs/server/MySQLHandler.cpp
View File

@ -2,7 +2,6 @@
#include <limits>
#include <ext/scope_guard.h>
#include <openssl/rsa.h>
#include <Columns/ColumnVector.h>
#include <Common/config_version.h>
#include <Common/NetException.h>
@ -45,6 +44,7 @@ MySQLHandler::MySQLHandler(IServer & server_, const Poco::Net::StreamSocket & so
, connection_id(connection_id_)
, public_key(public_key_)
, private_key(private_key_)
, auth_plugin(new Authentication::Native41())
{
server_capability_flags = CLIENT_PROTOCOL_41 | CLIENT_SECURE_CONNECTION | CLIENT_PLUGIN_AUTH | CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA | CLIENT_CONNECT_WITH_DB | CLIENT_DEPRECATE_EOF;
if (ssl_enabled)
@ -62,9 +62,7 @@ void MySQLHandler::run()
try
{
String scramble = generateScramble();
Handshake handshake(server_capability_flags, connection_id, VERSION_STRING + String("-") + VERSION_NAME, Authentication::Native, scramble + '\0');
Handshake handshake(server_capability_flags, connection_id, VERSION_STRING + String("-") + VERSION_NAME, auth_plugin->getName(), auth_plugin->getAuthPluginData());
packet_sender->sendPacket<Handshake>(handshake, true);
LOG_TRACE(log, "Sent handshake");
@ -96,10 +94,21 @@ void MySQLHandler::run()
client_capability_flags = handshake_response.capability_flags;
if (!(client_capability_flags & CLIENT_PROTOCOL_41))
throw Exception("Required capability: CLIENT_PROTOCOL_41.", ErrorCodes::MYSQL_CLIENT_INSUFFICIENT_CAPABILITIES);
if (!(client_capability_flags & CLIENT_PLUGIN_AUTH))
throw Exception("Required capability: CLIENT_PLUGIN_AUTH.", ErrorCodes::MYSQL_CLIENT_INSUFFICIENT_CAPABILITIES);
authenticate(handshake_response, scramble);
authenticate(handshake_response.username, handshake_response.auth_plugin_name, handshake_response.auth_response);
try
{
if (!handshake_response.database.empty())
connection_context.setCurrentDatabase(handshake_response.database);
connection_context.setCurrentQueryId("");
}
catch (const Exception & exc)
{
log->log(exc);
packet_sender->sendPacket(ERR_Packet(exc.code(), "00000", exc.message()), true);
}
OK_Packet ok_packet(0, handshake_response.capability_flags, 0, 0, 0);
packet_sender->sendPacket(ok_packet, true);
@ -216,121 +225,24 @@ void MySQLHandler::finishHandshake(MySQLProtocol::HandshakeResponse & packet)
}
}
String MySQLHandler::generateScramble()
void MySQLHandler::authenticate(const String & user_name, const String & auth_plugin_name, const String & initial_auth_response)
{
String scramble(MySQLProtocol::SCRAMBLE_LENGTH, 0);
Poco::RandomInputStream generator;
for (size_t i = 0; i < scramble.size(); i++)
{
generator >> scramble[i];
}
return scramble;
}
// For compatibility with JavaScript MySQL client, Native41 authentication plugin is used when possible (if password is specified using double SHA1). Otherwise SHA256 plugin is used.
auto user = connection_context.getUser(user_name);
if (user->password_double_sha1_hex.empty())
auth_plugin = std::make_unique<Authentication::Sha256Password>(public_key, private_key, log);
void MySQLHandler::authenticate(const HandshakeResponse & handshake_response, const String & scramble)
{
String auth_response;
AuthSwitchResponse response;
if (handshake_response.auth_plugin_name != Authentication::SHA256)
{
/** Native authentication sent 20 bytes + '\0' character = 21 bytes.
* This plugin must do the same to stay consistent with historical behavior if it is set to operate as a default plugin.
* https://github.com/mysql/mysql-server/blob/8.0/sql/auth/sql_authentication.cc#L3994
*/
packet_sender->sendPacket(AuthSwitchRequest(Authentication::SHA256, scramble + '\0'), true);
if (in->eof())
throw Exception(
"Client doesn't support authentication method " + String(Authentication::SHA256) + " used by ClickHouse",
ErrorCodes::MYSQL_CLIENT_INSUFFICIENT_CAPABILITIES);
packet_sender->receivePacket(response);
auth_response = response.value;
LOG_TRACE(log, "Authentication method mismatch.");
}
else
{
auth_response = handshake_response.auth_response;
LOG_TRACE(log, "Authentication method match.");
}
if (auth_response == "\1")
{
LOG_TRACE(log, "Client requests public key.");
BIO * mem = BIO_new(BIO_s_mem());
SCOPE_EXIT(BIO_free(mem));
if (PEM_write_bio_RSA_PUBKEY(mem, &public_key) != 1)
{
throw Exception("Failed to write public key to memory. Error: " + getOpenSSLErrors(), ErrorCodes::OPENSSL_ERROR);
}
char * pem_buf = nullptr;
long pem_size = BIO_get_mem_data(mem, &pem_buf);
String pem(pem_buf, pem_size);
LOG_TRACE(log, "Key: " << pem);
AuthMoreData data(pem);
packet_sender->sendPacket(data, true);
packet_sender->receivePacket(response);
auth_response = response.value;
}
else
{
LOG_TRACE(log, "Client didn't request public key.");
}
String password;
/** Decrypt password, if it's not empty.
* The original intention was that the password is a string[NUL] but this never got enforced properly so now we have to accept that
* an empty packet is a blank password, thus the check for auth_response.empty() has to be made too.
* https://github.com/mysql/mysql-server/blob/8.0/sql/auth/sql_authentication.cc#L4017
*/
if (!secure_connection && !auth_response.empty() && auth_response != String("\0", 1))
{
LOG_TRACE(log, "Received nonempty password");
auto ciphertext = reinterpret_cast<unsigned char *>(auth_response.data());
unsigned char plaintext[RSA_size(&private_key)];
int plaintext_size = RSA_private_decrypt(auth_response.size(), ciphertext, plaintext, &private_key, RSA_PKCS1_OAEP_PADDING);
if (plaintext_size == -1)
{
throw Exception("Failed to decrypt auth data. Error: " + getOpenSSLErrors(), ErrorCodes::OPENSSL_ERROR);
}
password.resize(plaintext_size);
for (int i = 0; i < plaintext_size; ++i)
{
password[i] = plaintext[i] ^ static_cast<unsigned char>(scramble[i % scramble.size()]);
}
}
else if (secure_connection)
{
password = auth_response;
}
else
{
LOG_TRACE(log, "Received empty password");
}
if (!password.empty() && password.back() == 0)
{
password.pop_back();
}
try
{
connection_context.setUser(handshake_response.username, password, socket().address(), "");
if (!handshake_response.database.empty()) connection_context.setCurrentDatabase(handshake_response.database);
connection_context.setCurrentQueryId("");
LOG_INFO(log, "Authentication for user " << handshake_response.username << " succeeded.");
try {
std::optional<String> auth_response = auth_plugin_name == auth_plugin->getName() ? std::make_optional<String>(initial_auth_response) : std::nullopt;
auth_plugin->authenticate(user_name, auth_response, connection_context, packet_sender, secure_connection, socket().address());
}
catch (const Exception & exc)
{
LOG_ERROR(log, "Authentication for user " << handshake_response.username << " failed.");
LOG_ERROR(log, "Authentication for user " << user_name << " failed.");
packet_sender->sendPacket(ERR_Packet(exc.code(), "00000", exc.message()), true);
throw;
}
LOG_INFO(log, "Authentication for user " << user_name << " succeeded.");
}
void MySQLHandler::comInitDB(ReadBuffer & payload)

6
dbms/programs/server/MySQLHandler.h
View File

@ -30,9 +30,7 @@ private:
void comInitDB(ReadBuffer & payload);
static String generateScramble();
void authenticate(const MySQLProtocol::HandshakeResponse &, const String & scramble);
void authenticate(const String & user_name, const String & auth_plugin_name, const String & auth_response);
IServer & server;
Poco::Logger * log;
@ -48,6 +46,8 @@ private:
RSA & public_key;
RSA & private_key;
std::unique_ptr<MySQLProtocol::Authentication::IPlugin> auth_plugin;
std::shared_ptr<Poco::Net::SecureStreamSocket> ss;
std::shared_ptr<ReadBuffer> in;
std::shared_ptr<WriteBuffer> out;

8
dbms/programs/server/users.xml
View File

@ -39,10 +39,18 @@
If you want to specify SHA256, place it in 'password_sha256_hex' element.
Example: <password_sha256_hex>65e84be33532fb784c48129675f9eff3a682b27168c0ea744b2cf58ee02337c5</password_sha256_hex>
Restrictions of SHA256: impossibility to connect to ClickHouse using MySQL JS client (as of July 2019).
If you want to specify double SHA1, place it in 'password_double_sha1_hex' element.
Example: <password_double_sha1_hex>e395796d6546b1b65db9d665cd43f0e858dd4303</password_double_sha1_hex>
How to generate decent password:
Execute: PASSWORD=$(base64 < /dev/urandom | head -c8); echo "$PASSWORD"; echo -n "$PASSWORD" | sha256sum | tr -d '-'
In first line will be password and in second - corresponding SHA256.
How to generate double SHA1:
Execute: PASSWORD=$(base64 < /dev/urandom | head -c8); echo "$PASSWORD"; echo -n "$PASSWORD" | openssl dgst -sha1 -binary | openssl dgst -sha1
In first line will be password and in second - corresponding double SHA1.
-->
<password></password>

2
dbms/src/Common/ProfileEvents.cpp
View File

@ -172,7 +172,7 @@
M(OSWriteChars, "Number of bytes written to filesystem, including page cache.") \
M(CreatedHTTPConnections, "Total amount of created HTTP connections (closed or opened).") \
\
M(QueryProfilerCannotWriteTrace, "Number of stack traces dropped by query profiler because pipe is full or cannot write to pipe.") \
M(CannotWriteToWriteBufferDiscard, "Number of stack traces dropped by query profiler or signal handler because pipe is full or cannot write to pipe.") \
M(QueryProfilerSignalOverruns, "Number of times we drop processing of a signal due to overrun plus the number of signals that OS has not delivered due to overrun.") \
namespace ProfileEvents

35
dbms/src/Common/QueryProfiler.cpp
View File

@ -11,12 +11,11 @@
#include <Common/Exception.h>
#include <Common/thread_local_rng.h>
#include <IO/WriteHelpers.h>
#include <IO/WriteBufferFromFileDescriptor.h>
#include <IO/WriteBufferFromFileDescriptorDiscardOnFailure.h>
namespace ProfileEvents
{
extern const Event QueryProfilerCannotWriteTrace;
extern const Event QueryProfilerSignalOverruns;
}
@ -27,36 +26,6 @@ extern LazyPipe trace_pipe;
namespace
{
/** Write to file descriptor but drop the data if write would block or fail.
* To use within signal handler. Motivating example: a signal handler invoked during execution of malloc
* should not block because some mutex (or even worse - a spinlock) may be held.
*/
class WriteBufferDiscardOnFailure : public WriteBufferFromFileDescriptor
{
protected:
void nextImpl() override
{
size_t bytes_written = 0;
while (bytes_written != offset())
{
ssize_t res = ::write(fd, working_buffer.begin() + bytes_written, offset() - bytes_written);
if ((-1 == res || 0 == res) && errno != EINTR)
{
ProfileEvents::increment(ProfileEvents::QueryProfilerCannotWriteTrace);
break; /// Discard
}
if (res > 0)
bytes_written += res;
}
}
public:
using WriteBufferFromFileDescriptor::WriteBufferFromFileDescriptor;
~WriteBufferDiscardOnFailure() override {}
};
/// Normally query_id is a UUID (string with a fixed length) but user can provide custom query_id.
/// Thus upper bound on query_id length should be introduced to avoid buffer overflow in signal handler.
constexpr size_t QUERY_ID_MAX_LEN = 1024;
@ -90,7 +59,7 @@ namespace
sizeof(TimerType) + // timer type
sizeof(UInt32); // thread_number
char buffer[buf_size];
WriteBufferDiscardOnFailure out(trace_pipe.fds_rw[1], buf_size, buffer);
WriteBufferFromFileDescriptorDiscardOnFailure out(trace_pipe.fds_rw[1], buf_size, buffer);
StringRef query_id = CurrentThread::getQueryId();
query_id.size = std::min(query_id.size, QUERY_ID_MAX_LEN);

23
dbms/src/Common/StackTrace.cpp
View File

@ -151,6 +151,12 @@ std::string signalToErrorMessage(int sig, const siginfo_t & info, const ucontext
}
break;
}
case SIGPROF:
{
error << "This is a signal used for debugging purposes by the user.";
break;
}
}
return error.str();
@ -239,10 +245,10 @@ const StackTrace::Frames & StackTrace::getFrames() const
}
static std::string toStringImpl(const StackTrace::Frames & frames, size_t offset, size_t size)
static void toStringEveryLineImpl(const StackTrace::Frames & frames, size_t offset, size_t size, std::function<void(const std::string &)> callback)
{
if (size == 0)
return "<Empty trace>";
return callback("<Empty trace>");
const DB::SymbolIndex & symbol_index = DB::SymbolIndex::instance();
std::unordered_map<std::string, DB::Dwarf> dwarfs;
@ -281,12 +287,23 @@ static std::string toStringImpl(const StackTrace::Frames & frames, size_t offset
else
out << "?";
out << "\n";
callback(out.str());
out.str({});
}
}
static std::string toStringImpl(const StackTrace::Frames & frames, size_t offset, size_t size)
{
std::stringstream out;
toStringEveryLineImpl(frames, offset, size, [&](const std::string & str) { out << str << '\n'; });
return out.str();
}
void StackTrace::toStringEveryLine(std::function<void(const std::string &)> callback) const
{
toStringEveryLineImpl(frames, offset, size, std::move(callback));
}
std::string StackTrace::toString() const
{
/// Calculation of stack trace text is extremely slow.

3
dbms/src/Common/StackTrace.h
View File

@ -4,6 +4,7 @@
#include <vector>
#include <array>
#include <optional>
#include <functional>
#include <signal.h>
#ifdef __APPLE__
@ -39,6 +40,8 @@ public:
const Frames & getFrames() const;
std::string toString() const;
void toStringEveryLine(std::function<void(const std::string &)> callback) const;
protected:
void tryCapture();

262
dbms/src/Core/MySQLProtocol.h
View File

@ -1,9 +1,17 @@
#pragma once
#include <ext/scope_guard.h>
#include <openssl/pem.h>
#include <openssl/rsa.h>
#include <random>
#include <sstream>
#include <Common/MemoryTracker.h>
#include <Common/OpenSSLHelpers.h>
#include <Common/PODArray.h>
#include <Core/Types.h>
#include <Interpreters/Context.h>
#include <IO/copyData.h>
#include <IO/LimitReadBuffer.h>
#include <IO/ReadBuffer.h>
#include <IO/ReadBufferFromMemory.h>
#include <IO/ReadBufferFromPocoSocket.h>
@ -14,9 +22,7 @@
#include <IO/WriteHelpers.h>
#include <Poco/Net/StreamSocket.h>
#include <Poco/RandomStream.h>
#include <random>
#include <sstream>
#include <IO/LimitReadBuffer.h>
#include <Poco/SHA1Engine.h>
/// Implementation of MySQL wire protocol.
/// Works only on little-endian architecture.
@ -27,6 +33,9 @@ namespace DB
namespace ErrorCodes
{
extern const int UNKNOWN_PACKET_FROM_CLIENT;
extern const int MYSQL_CLIENT_INSUFFICIENT_CAPABILITIES;
extern const int OPENSSL_ERROR;
extern const int UNKNOWN_EXCEPTION;
}
namespace MySQLProtocol
@ -39,11 +48,6 @@ const size_t MYSQL_ERRMSG_SIZE = 512;
const size_t PACKET_HEADER_SIZE = 4;
const size_t SSL_REQUEST_PAYLOAD_SIZE = 32;
namespace Authentication
{
const String Native = "mysql_native_password";
const String SHA256 = "sha256_password"; /// Caching SHA2 plugin is not used because it would be possible to authenticate knowing hash from users.xml.
}
enum CharacterSet
{
@ -149,6 +153,8 @@ private:
uint8_t & sequence_id;
const size_t max_packet_size = MAX_PACKET_LENGTH;
bool has_read_header = false;
// Size of packet which is being read now.
size_t payload_length = 0;
@ -158,8 +164,9 @@ private:
protected:
bool nextImpl() override
{
if (payload_length == 0 || (payload_length == max_packet_size && offset == payload_length))
if (!has_read_header || (payload_length == max_packet_size && offset == payload_length))
{
has_read_header = true;
working_buffer.resize(0);
offset = 0;
payload_length = 0;
@ -171,10 +178,6 @@ protected:
tmp << "Received packet with payload larger than max_packet_size: " << payload_length;
throw ProtocolError(tmp.str(), ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT);
}
else if (payload_length == 0)
{
return false;
}
size_t packet_sequence_id = 0;
in.read(reinterpret_cast<char &>(packet_sequence_id));
@ -185,6 +188,9 @@ protected:
throw ProtocolError(tmp.str(), ErrorCodes::UNKNOWN_PACKET_FROM_CLIENT);
}
sequence_id++;
if (payload_length == 0)
return false;
}
else if (offset == payload_length)
{
@ -208,6 +214,7 @@ class ClientPacket
{
public:
ClientPacket() = default;
ClientPacket(ClientPacket &&) = default;
virtual void read(ReadBuffer & in, uint8_t & sequence_id)
@ -257,6 +264,7 @@ public:
{
return total_left;
}
private:
WriteBuffer & out;
uint8_t & sequence_id;
@ -452,9 +460,6 @@ protected:
buffer.write(static_cast<char>(auth_plugin_data.size()));
writeChar(0x0, 10, buffer);
writeString(auth_plugin_data.substr(AUTH_PLUGIN_DATA_PART_1_LENGTH, auth_plugin_data.size() - AUTH_PLUGIN_DATA_PART_1_LENGTH), buffer);
// A workaround for PHP mysqlnd extension bug which occurs when sha256_password is used as a default authentication plugin.
// Instead of using client response for mysql_native_password plugin, the server will always generate authentication method mismatch
// and switch to sha256_password to simulate that mysql_native_password is used as a default plugin.
writeString(auth_plugin_name, buffer);
writeChar(0x0, 1, buffer);
}
@ -843,5 +848,230 @@ protected:
}
};
namespace Authentication
{
class IPlugin
{
public:
virtual String getName() = 0;
virtual String getAuthPluginData() = 0;
virtual void authenticate(const String & user_name, std::optional<String> auth_response, Context & context, std::shared_ptr<PacketSender> packet_sender, bool is_secure_connection,
const Poco::Net::SocketAddress & address) = 0;
virtual ~IPlugin() = default;
};
/// https://dev.mysql.com/doc/internals/en/secure-password-authentication.html
class Native41 : public IPlugin
{
public:
Native41()
{
scramble.resize(SCRAMBLE_LENGTH + 1, 0);
Poco::RandomInputStream generator;
for (size_t i = 0; i < SCRAMBLE_LENGTH; i++)
generator >> scramble[i];
}
String getName() override
{
return "mysql_native_password";
}
String getAuthPluginData() override
{
return scramble;
}
void authenticate(
const String & user_name,
std::optional<String> auth_response,
Context & context,
std::shared_ptr<PacketSender> packet_sender,
bool /* is_secure_connection */,
const Poco::Net::SocketAddress & address) override
{
if (!auth_response)
{
packet_sender->sendPacket(AuthSwitchRequest(getName(), scramble), true);
AuthSwitchResponse response;
packet_sender->receivePacket(response);
auth_response = response.value;
}
if (auth_response->empty())
{
context.setUser(user_name, "", address, "");
return;
}
if (auth_response->size() != Poco::SHA1Engine::DIGEST_SIZE)
throw Exception("Wrong size of auth response. Expected: " + std::to_string(Poco::SHA1Engine::DIGEST_SIZE) + " bytes, received: " + std::to_string(auth_response->size()) + " bytes.",
ErrorCodes::UNKNOWN_EXCEPTION);
auto user = context.getUser(user_name);
if (user->password_double_sha1_hex.empty())
throw Exception("Cannot use " + getName() + " auth plugin for user " + user_name + " since its password isn't specified using double SHA1.", ErrorCodes::UNKNOWN_EXCEPTION);
Poco::SHA1Engine::Digest double_sha1_value = Poco::DigestEngine::digestFromHex(user->password_double_sha1_hex);
assert(double_sha1_value.size() == Poco::SHA1Engine::DIGEST_SIZE);
Poco::SHA1Engine engine;
engine.update(scramble.data(), SCRAMBLE_LENGTH);
engine.update(double_sha1_value.data(), double_sha1_value.size());
String password_sha1(Poco::SHA1Engine::DIGEST_SIZE, 0x0);
const Poco::SHA1Engine::Digest & digest = engine.digest();
for (size_t i = 0; i < password_sha1.size(); i++)
{
password_sha1[i] = digest[i] ^ static_cast<unsigned char>((*auth_response)[i]);
}
context.setUser(user_name, password_sha1, address, "");
}
private:
String scramble;
};
/// Caching SHA2 plugin is not used because it would be possible to authenticate knowing hash from users.xml.
/// https://dev.mysql.com/doc/internals/en/sha256.html
class Sha256Password : public IPlugin
{
public:
Sha256Password(RSA & public_key_, RSA & private_key_, Logger * log_)
: public_key(public_key_)
, private_key(private_key_)
, log(log_)
{
/** Native authentication sent 20 bytes + '\0' character = 21 bytes.
* This plugin must do the same to stay consistent with historical behavior if it is set to operate as a default plugin. [1]
* https://github.com/mysql/mysql-server/blob/8.0/sql/auth/sql_authentication.cc#L3994
*/
scramble.resize(SCRAMBLE_LENGTH + 1, 0);
Poco::RandomInputStream generator;
for (size_t i = 0; i < SCRAMBLE_LENGTH; i++)
generator >> scramble[i];
}
String getName() override
{
return "sha256_password";
}
String getAuthPluginData() override
{
return scramble;
}
void authenticate(
const String & user_name,
std::optional<String> auth_response,
Context & context,
std::shared_ptr<PacketSender> packet_sender,
bool is_secure_connection,
const Poco::Net::SocketAddress & address) override
{
if (!auth_response)
{
packet_sender->sendPacket(AuthSwitchRequest(getName(), scramble), true);
if (packet_sender->in->eof())
throw Exception("Client doesn't support authentication method " + getName() + " used by ClickHouse. Specifying user password using 'password_double_sha1_hex' may fix the problem.",
ErrorCodes::MYSQL_CLIENT_INSUFFICIENT_CAPABILITIES);
AuthSwitchResponse response;
packet_sender->receivePacket(response);
auth_response = response.value;
LOG_TRACE(log, "Authentication method mismatch.");
}
else
{
LOG_TRACE(log, "Authentication method match.");
}
if (auth_response == "\1")
{
LOG_TRACE(log, "Client requests public key.");
BIO * mem = BIO_new(BIO_s_mem());
SCOPE_EXIT(BIO_free(mem));
if (PEM_write_bio_RSA_PUBKEY(mem, &public_key) != 1)
{
throw Exception("Failed to write public key to memory. Error: " + getOpenSSLErrors(), ErrorCodes::OPENSSL_ERROR);
}
char * pem_buf = nullptr;
long pem_size = BIO_get_mem_data(mem, &pem_buf);
String pem(pem_buf, pem_size);
LOG_TRACE(log, "Key: " << pem);
AuthMoreData data(pem);
packet_sender->sendPacket(data, true);
AuthSwitchResponse response;
packet_sender->receivePacket(response);
auth_response = response.value;
}
else
{
LOG_TRACE(log, "Client didn't request public key.");
}
String password;
/** Decrypt password, if it's not empty.
* The original intention was that the password is a string[NUL] but this never got enforced properly so now we have to accept that
* an empty packet is a blank password, thus the check for auth_response.empty() has to be made too.
* https://github.com/mysql/mysql-server/blob/8.0/sql/auth/sql_authentication.cc#L4017
*/
if (!is_secure_connection && !auth_response->empty() && auth_response != String("\0", 1))
{
LOG_TRACE(log, "Received nonempty password");
auto ciphertext = reinterpret_cast<unsigned char *>(auth_response->data());
unsigned char plaintext[RSA_size(&private_key)];
int plaintext_size = RSA_private_decrypt(auth_response->size(), ciphertext, plaintext, &private_key, RSA_PKCS1_OAEP_PADDING);
if (plaintext_size == -1)
{
throw Exception("Failed to decrypt auth data. Error: " + getOpenSSLErrors(), ErrorCodes::OPENSSL_ERROR);
}
password.resize(plaintext_size);
for (int i = 0; i < plaintext_size; i++)
{
password[i] = plaintext[i] ^ static_cast<unsigned char>(scramble[i % scramble.size()]);
}
}
else if (is_secure_connection)
{
password = *auth_response;
}
else
{
LOG_TRACE(log, "Received empty password");
}
if (!password.empty() && password.back() == 0)
{
password.pop_back();
}
context.setUser(user_name, password, address, "");
}
private:
RSA & public_key;
RSA & private_key;
Logger * log;
String scramble;
};
}
}
}

29
dbms/src/IO/WriteBufferFromFileDescriptorDiscardOnFailure.cpp Normal file
View File

@ -0,0 +1,29 @@
#include <IO/WriteBufferFromFileDescriptorDiscardOnFailure.h>
namespace ProfileEvents
{
extern const Event CannotWriteToWriteBufferDiscard;
}
namespace DB
{
void WriteBufferFromFileDescriptorDiscardOnFailure::nextImpl()
{
size_t bytes_written = 0;
while (bytes_written != offset())
{
ssize_t res = ::write(fd, working_buffer.begin() + bytes_written, offset() - bytes_written);
if ((-1 == res || 0 == res) && errno != EINTR)
{
ProfileEvents::increment(ProfileEvents::CannotWriteToWriteBufferDiscard);
break; /// Discard
}
if (res > 0)
bytes_written += res;
}
}
}

23
dbms/src/IO/WriteBufferFromFileDescriptorDiscardOnFailure.h Normal file
View File

@ -0,0 +1,23 @@
#pragma once
#include <IO/WriteBufferFromFileDescriptor.h>
namespace DB
{
/** Write to file descriptor but drop the data if write would block or fail.
* To use within signal handler. Motivating example: a signal handler invoked during execution of malloc
* should not block because some mutex (or even worse - a spinlock) may be held.
*/
class WriteBufferFromFileDescriptorDiscardOnFailure : public WriteBufferFromFileDescriptor
{
protected:
void nextImpl() override;
public:
using WriteBufferFromFileDescriptor::WriteBufferFromFileDescriptor;
~WriteBufferFromFileDescriptorDiscardOnFailure() override {}
};
}

25
dbms/src/Interpreters/AsynchronousMetrics.cpp
View File

@ -180,19 +180,22 @@ void AsynchronousMetrics::update()
calculateMaxAndSum(max_inserts_in_queue, sum_inserts_in_queue, status.queue.inserts_in_queue);
calculateMaxAndSum(max_merges_in_queue, sum_merges_in_queue, status.queue.merges_in_queue);
try
if (!status.is_readonly)
{
time_t absolute_delay = 0;
time_t relative_delay = 0;
table_replicated_merge_tree->getReplicaDelays(absolute_delay, relative_delay);
try
{
time_t absolute_delay = 0;
time_t relative_delay = 0;
table_replicated_merge_tree->getReplicaDelays(absolute_delay, relative_delay);
calculateMax(max_absolute_delay, absolute_delay);
calculateMax(max_relative_delay, relative_delay);
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__,
"Cannot get replica delay for table: " + backQuoteIfNeed(db.first) + "." + backQuoteIfNeed(iterator->name()));
calculateMax(max_absolute_delay, absolute_delay);
calculateMax(max_relative_delay, relative_delay);
}
catch (...)
{
tryLogCurrentException(__PRETTY_FUNCTION__,
"Cannot get replica delay for table: " + backQuoteIfNeed(db.first) + "." + backQuoteIfNeed(iterator->name()));
}
}
calculateMax(max_part_count_for_partition, table_replicated_merge_tree->getMaxPartsCountForPartition());

4
dbms/src/Interpreters/Context.cpp
View File

@ -655,6 +655,10 @@ void Context::setProfile(const String & profile)
settings_constraints = std::move(new_constraints);
}
std::shared_ptr<const User> Context::getUser(const String & user_name)
{
return shared->users_manager->getUser(user_name);
}
void Context::setUser(const String & name, const String & password, const Poco::Net::SocketAddress & address, const String & quota_key)
{

5
dbms/src/Interpreters/Context.h
View File

@ -6,6 +6,7 @@
#include <Core/Types.h>
#include <DataStreams/IBlockStream_fwd.h>
#include <Interpreters/ClientInfo.h>
#include <Interpreters/Users.h>
#include <Parsers/IAST_fwd.h>
#include <Common/LRUCache.h>
#include <Common/MultiVersion.h>
@ -200,6 +201,10 @@ public:
/// Must be called before getClientInfo.
void setUser(const String & name, const String & password, const Poco::Net::SocketAddress & address, const String & quota_key);
/// Used by MySQL Secure Password Authentication plugin.
std::shared_ptr<const User> getUser(const String & user_name);
/// Compute and set actual user settings, client_info.current_user should be set
void calculateUserSettings();

58
dbms/src/Interpreters/MetricLog.cpp
View File

@ -1,9 +1,9 @@
#include <Interpreters/MetricLog.h>
#include <DataTypes/DataTypesNumber.h>
#include <DataTypes/DataTypeString.h>
#include <DataTypes/DataTypeDate.h>
#include <DataTypes/DataTypeDateTime.h>
namespace DB
{
@ -11,11 +11,10 @@ Block MetricLogElement::createBlock()
{
ColumnsWithTypeAndName columns_with_type_and_name;
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeDate>(), "event_date");
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeDateTime>(), "event_time");
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeUInt64>(), "milliseconds");
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeDate>(), "event_date");
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeDateTime>(), "event_time");
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeUInt64>(), "milliseconds");
//ProfileEvents
for (size_t i = 0, end = ProfileEvents::end(); i < end; ++i)
{
std::string name;
@ -24,7 +23,6 @@ Block MetricLogElement::createBlock()
columns_with_type_and_name.emplace_back(std::make_shared<DataTypeUInt64>(), std::move(name));
}
//CurrentMetrics
for (size_t i = 0, end = CurrentMetrics::end(); i < end; ++i)
{
std::string name;
@ -36,31 +34,25 @@ Block MetricLogElement::createBlock()
return Block(columns_with_type_and_name);
}
void MetricLogElement::appendToBlock(Block & block) const
{
MutableColumns columns = block.mutateColumns();
size_t iter = 0;
size_t column_idx = 0;
columns[iter++]->insert(DateLUT::instance().toDayNum(event_time));
columns[iter++]->insert(event_time);
columns[iter++]->insert(milliseconds);
columns[column_idx++]->insert(DateLUT::instance().toDayNum(event_time));
columns[column_idx++]->insert(event_time);
columns[column_idx++]->insert(milliseconds);
//ProfileEvents
for (size_t i = 0, end = ProfileEvents::end(); i < end; ++i)
{
const UInt64 value = ProfileEvents::global_counters[i].load(std::memory_order_relaxed);
columns[iter++]->insert(value);
}
columns[column_idx++]->insert(profile_events[i]);
//CurrentMetrics
for (size_t i = 0, end = CurrentMetrics::end(); i < end; ++i)
{
const UInt64 value = CurrentMetrics::values[i];
columns[iter++]->insert(value);
}
columns[column_idx++]->insert(current_metrics[i]);
}
void MetricLog::startCollectMetric(size_t collect_interval_milliseconds_)
{
collect_interval_milliseconds = collect_interval_milliseconds_;
@ -68,6 +60,7 @@ void MetricLog::startCollectMetric(size_t collect_interval_milliseconds_)
metric_flush_thread = ThreadFromGlobalPool([this] { metricThreadFunction(); });
}
void MetricLog::stopCollectMetric()
{
bool old_val = false;
@ -76,28 +69,51 @@ void MetricLog::stopCollectMetric()
metric_flush_thread.join();
}
inline UInt64 time_in_milliseconds(std::chrono::time_point<std::chrono::system_clock> timepoint)
{
return std::chrono::duration_cast<std::chrono::milliseconds>(timepoint.time_since_epoch()).count();
}
inline UInt64 time_in_seconds(std::chrono::time_point<std::chrono::system_clock> timepoint)
{
return std::chrono::duration_cast<std::chrono::seconds>(timepoint.time_since_epoch()).count();
}
void MetricLog::metricThreadFunction()
{
auto desired_timepoint = std::chrono::system_clock::now();
/// For differentiation of ProfileEvents counters.
std::vector<ProfileEvents::Count> prev_profile_events(ProfileEvents::end());
while (!is_shutdown_metric_thread)
{
try
{
MetricLogElement elem;
const auto current_time = std::chrono::system_clock::now();
MetricLogElement elem;
elem.event_time = std::chrono::system_clock::to_time_t(current_time);
elem.milliseconds = time_in_milliseconds(current_time) - time_in_seconds(current_time) * 1000;
elem.profile_events.resize(ProfileEvents::end());
for (size_t i = 0, end = ProfileEvents::end(); i < end; ++i)
{
const ProfileEvents::Count new_value = ProfileEvents::global_counters[i].load(std::memory_order_relaxed);
UInt64 & old_value = prev_profile_events[i];
elem.profile_events[i] = new_value - old_value;
old_value = new_value;
}
elem.current_metrics.resize(CurrentMetrics::end());
for (size_t i = 0, end = CurrentMetrics::end(); i < end; ++i)
{
elem.current_metrics[i] = CurrentMetrics::values[i];
}
this->add(elem);
/// We will record current time into table but align it to regular time intervals to avoid time drift.

16
dbms/src/Interpreters/MetricLog.h
View File

@ -1,22 +1,34 @@
#pragma once
#include <Interpreters/SystemLog.h>
#include <Interpreters/AsynchronousMetrics.h>
#include <Common/ProfileEvents.h>
#include <Common/CurrentMetrics.h>
#include <vector>
#include <atomic>
#include <ctime>
namespace DB
{
using Poco::Message;
/** MetricLog is a log of metric values measured at regular time interval.
*/
struct MetricLogElement
{
time_t event_time{};
UInt64 milliseconds{};
std::vector<ProfileEvents::Count> profile_events;
std::vector<CurrentMetrics::Metric> current_metrics;
static std::string name() { return "MetricLog"; }
static Block createBlock();
void appendToBlock(Block & block) const;
};
class MetricLog : public SystemLog<MetricLogElement>
{
using SystemLog<MetricLogElement>::SystemLog;

18
dbms/src/Interpreters/Users.cpp
View File

@ -278,12 +278,14 @@ User::User(const String & name_, const String & config_elem, const Poco::Util::A
{
bool has_password = config.has(config_elem + ".password");
bool has_password_sha256_hex = config.has(config_elem + ".password_sha256_hex");
bool has_password_double_sha1_hex = config.has(config_elem + ".password_double_sha1_hex");
if (has_password && has_password_sha256_hex)
throw Exception("Both fields 'password' and 'password_sha256_hex' are specified for user " + name + ". Must be only one of them.", ErrorCodes::BAD_ARGUMENTS);
if (has_password + has_password_sha256_hex + has_password_double_sha1_hex > 1)
throw Exception("More than one field of 'password', 'password_sha256_hex', 'password_double_sha1_hex' is used to specify password for user " + name + ". Must be only one of them.",
ErrorCodes::BAD_ARGUMENTS);
if (!has_password && !has_password_sha256_hex)
throw Exception("Either 'password' or 'password_sha256_hex' must be specified for user " + name + ".", ErrorCodes::BAD_ARGUMENTS);
if (!has_password && !has_password_sha256_hex && !has_password_double_sha1_hex)
throw Exception("Either 'password' or 'password_sha256_hex' or 'password_double_sha1_hex' must be specified for user " + name + ".", ErrorCodes::BAD_ARGUMENTS);
if (has_password)
password = config.getString(config_elem + ".password");
@ -296,6 +298,14 @@ User::User(const String & name_, const String & config_elem, const Poco::Util::A
throw Exception("password_sha256_hex for user " + name + " has length " + toString(password_sha256_hex.size()) + " but must be exactly 64 symbols.", ErrorCodes::BAD_ARGUMENTS);
}
if (has_password_double_sha1_hex)
{
password_double_sha1_hex = Poco::toLower(config.getString(config_elem + ".password_double_sha1_hex"));
if (password_double_sha1_hex.size() != 40)
throw Exception("password_double_sha1_hex for user " + name + " has length " + toString(password_double_sha1_hex.size()) + " but must be exactly 40 symbols.", ErrorCodes::BAD_ARGUMENTS);
}
profile = config.getString(config_elem + ".profile");
quota = config.getString(config_elem + ".quota");

1
dbms/src/Interpreters/Users.h
View File

@ -56,6 +56,7 @@ struct User
/// Required password. Could be stored in plaintext or in SHA256.
String password;
String password_sha256_hex;
String password_double_sha1_hex;
String profile;
String quota;

26
dbms/src/Interpreters/UsersManager.cpp
View File

@ -1,14 +1,15 @@
#include <Interpreters/UsersManager.h>
#include <Poco/Net/IPAddress.h>
#include <Poco/Util/AbstractConfiguration.h>
#include <Poco/String.h>
#include "config_core.h"
#include <Common/Exception.h>
#include <common/logger_useful.h>
#include <IO/HexWriteBuffer.h>
#include <IO/WriteBufferFromString.h>
#include <IO/WriteHelpers.h>
#include <common/logger_useful.h>
#include "config_core.h"
#include <Poco/Net/IPAddress.h>
#include <Poco/SHA1Engine.h>
#include <Poco/String.h>
#include <Poco/Util/AbstractConfiguration.h>
#if USE_SSL
# include <openssl/sha.h>
#endif
@ -93,6 +94,21 @@ UserPtr UsersManager::authorizeAndGetUser(
throw DB::Exception("SHA256 passwords support is disabled, because ClickHouse was built without SSL library", DB::ErrorCodes::SUPPORT_IS_DISABLED);
#endif
}
else if (!it->second->password_double_sha1_hex.empty())
{
Poco::SHA1Engine engine;
engine.update(password);
const auto & first_sha1 = engine.digest();
/// If it was MySQL compatibility server, then first_sha1 already contains double SHA1.
if (Poco::SHA1Engine::digestToHex(first_sha1) == it->second->password_double_sha1_hex)
return it->second;
engine.update(first_sha1.data(), first_sha1.size());
if (Poco::SHA1Engine::digestToHex(engine.digest()) != it->second->password_double_sha1_hex)
on_wrong_password();
}
else if (password != it->second->password)
{
on_wrong_password();

5
dbms/tests/integration/test_mysql_protocol/clients/mysqljs/Dockerfile Normal file
View File

@ -0,0 +1,5 @@
FROM node:8
RUN npm install mysql
COPY ./test.js test.js

8
dbms/tests/integration/test_mysql_protocol/clients/mysqljs/docker_compose.yml Normal file
View File

@ -0,0 +1,8 @@
version: '2.2'
services:
mysqljs1:
build:
context: ./
network: host
# to keep container running
command: sleep infinity

21
dbms/tests/integration/test_mysql_protocol/clients/mysqljs/test.js Normal file
View File

@ -0,0 +1,21 @@
var mysql = require('mysql');
var connection = mysql.createConnection({
host : process.argv[2],
port : process.argv[3],
user : process.argv[4],
password : process.argv[5],
database : 'system',
});
connection.connect();
connection.query('SELECT 1 + 1 AS solution', function (error, results, fields) {
if (error) throw error;
if (results[0].solution.toString() !== '2') {
throw Error('Wrong result of a query. Expected: "2", received: ' + results[0].solution + '.')
}
});
connection.end();

20
dbms/tests/integration/test_mysql_protocol/configs/users.xml
View File

@ -14,6 +14,26 @@
<profile>default</profile>
<quota>default</quota>
</default>
<user_with_double_sha1>
<!-- echo -n abacaba | openssl dgst -sha1 -binary | openssl dgst -sha1 !-->
<password_double_sha1_hex>e395796d6546b1b65db9d665cd43f0e858dd4303</password_double_sha1_hex>
<networks incl="networks" replace="replace">
<ip>::/0</ip>
</networks>
<profile>default</profile>
<quota>default</quota>
</user_with_double_sha1>
<user_with_empty_password>
<password></password>
<networks incl="networks" replace="replace">
<ip>::/0</ip>
</networks>
<profile>default</profile>
<quota>default</quota>
</user_with_empty_password>
</users>
<quotas>

34
dbms/tests/integration/test_mysql_protocol/test.py
View File

@ -50,8 +50,22 @@ def php_container():
yield docker.from_env().containers.get(cluster.project_name + '_php1_1')
@pytest.fixture(scope='module')
def nodejs_container():
docker_compose = os.path.join(SCRIPT_DIR, 'clients', 'mysqljs', 'docker_compose.yml')
subprocess.check_call(['docker-compose', '-p', cluster.project_name, '-f', docker_compose, 'up', '--no-recreate', '-d', '--build'])
yield docker.from_env().containers.get(cluster.project_name + '_mysqljs1_1')
def test_mysql_client(mysql_client, server_address):
# type: (Container, str) -> None
code, (stdout, stderr) = mysql_client.exec_run('''
mysql --protocol tcp -h {host} -P {port} default -u user_with_double_sha1 --password=abacaba
-e "SELECT 1;"
'''.format(host=server_address, port=server_port), demux=True)
assert stdout == '\n'.join(['1', '1', ''])
code, (stdout, stderr) = mysql_client.exec_run('''
mysql --protocol tcp -h {host} -P {port} default -u default --password=123
-e "SELECT 1 as a;"
@ -149,10 +163,26 @@ def test_golang_client(server_address, golang_container):
def test_php_client(server_address, php_container):
# type: (str, Container) -> None
code, (stdout, stderr) = php_container.exec_run('php -f test.php {host} {port} default 123 '.format(host=server_address, port=server_port), demux=True)
code, (stdout, stderr) = php_container.exec_run('php -f test.php {host} {port} default 123'.format(host=server_address, port=server_port), demux=True)
assert code == 0
assert stdout == 'tables\n'
code, (stdout, stderr) = php_container.exec_run('php -f test_ssl.php {host} {port} default 123 '.format(host=server_address, port=server_port), demux=True)
code, (stdout, stderr) = php_container.exec_run('php -f test_ssl.php {host} {port} default 123'.format(host=server_address, port=server_port), demux=True)
assert code == 0
assert stdout == 'tables\n'
def test_mysqljs_client(server_address, nodejs_container):
code, (_, stderr) = nodejs_container.exec_run('node test.js {host} {port} default 123'.format(host=server_address, port=server_port), demux=True)
assert code == 1
assert 'MySQL is requesting the sha256_password authentication method, which is not supported.' in stderr
code, (_, stderr) = nodejs_container.exec_run('node test.js {host} {port} user_with_empty_password ""'.format(host=server_address, port=server_port), demux=True)
assert code == 1
assert 'MySQL is requesting the sha256_password authentication method, which is not supported.' in stderr
code, (_, _) = nodejs_container.exec_run('node test.js {host} {port} user_with_double_sha1 abacaba'.format(host=server_address, port=server_port), demux=True)
assert code == 0
code, (_, _) = nodejs_container.exec_run('node test.js {host} {port} user_with_empty_password 123'.format(host=server_address, port=server_port), demux=True)
assert code == 1

43
dbms/tests/performance/json_extract_rapidjson.xml
View File

@ -1,18 +1,19 @@
<test>
<type>once</type>
<type>loop</type>
<stop_conditions>
<all_of>
<iterations>3</iterations>
<min_time_not_changing_for_ms>10000</min_time_not_changing_for_ms>
</all_of>
<any_of>
<average_speed_not_changing_for_ms>1000</average_speed_not_changing_for_ms>
<total_time_ms>2000</total_time_ms>
<iterations>5</iterations>
<total_time_ms>60000</total_time_ms>
</any_of>
</stop_conditions>
<main_metric>
<max_rows_per_second />
<max_bytes_per_second />
<avg_rows_per_second />
<avg_bytes_per_second />
<min_time/>
</main_metric>
<substitutions>
@ -34,18 +35,18 @@
<allow_simdjson>0</allow_simdjson>
</settings>
<query>SELECT 'rapidjson-1', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam'))</query>
<query>SELECT 'rapidjson-2', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'rapidjson-3', count() FROM system.numbers WHERE NOT ignore(JSONExtractInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'rapidjson-4', count() FROM system.numbers WHERE NOT ignore(JSONExtractUInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'rapidjson-5', count() FROM system.numbers WHERE NOT ignore(JSONExtractFloat(materialize({json}), 'fparam'))</query>
<query>SELECT 'rapidjson-6', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam'))</query>
<query>SELECT 'rapidjson-7', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'rapidjson-8', count() FROM system.numbers WHERE NOT ignore(JSONExtractInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'rapidjson-9', count() FROM system.numbers WHERE NOT ignore(JSONExtractUInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'rapidjson-10', count() FROM system.numbers WHERE NOT ignore(JSONExtractRaw(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'rapidjson-11', count() FROM system.numbers WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'rapidjson-12', count() FROM system.numbers WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam', 'nested_2', -2))</query>
<query>SELECT 'rapidjson-13', count() FROM system.numbers WHERE NOT ignore(JSONExtractBool(materialize({long_json}), 'bparam'))</query>
<query>SELECT 'rapidjson-1', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam'))</query>
<query>SELECT 'rapidjson-2', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'rapidjson-3', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'rapidjson-4', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractUInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'rapidjson-5', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractFloat(materialize({json}), 'fparam'))</query>
<query>SELECT 'rapidjson-6', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam'))</query>
<query>SELECT 'rapidjson-7', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'rapidjson-8', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'rapidjson-9', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractUInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'rapidjson-10', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractRaw(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'rapidjson-11', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'rapidjson-12', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam', 'nested_2', -2))</query>
<query>SELECT 'rapidjson-13', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractBool(materialize({long_json}), 'bparam'))</query>
</test>

59
dbms/tests/performance/json_extract_simdjson.xml
View File

@ -1,23 +1,20 @@
<test>
<type>once</type>
<type>loop</type>
<preconditions>
<cpu>AVX2</cpu>
</preconditions>
<stop_conditions>
<all_of>
<iterations>3</iterations>
<min_time_not_changing_for_ms>10000</min_time_not_changing_for_ms>
</all_of>
<any_of>
<iterations>5</iterations>
<total_time_ms>60000</total_time_ms>
</any_of>
</stop_conditions>
<stop_conditions>
<any_of>
<average_speed_not_changing_for_ms>1000</average_speed_not_changing_for_ms>
<total_time_ms>2000</total_time_ms>
</any_of>
</stop_conditions>
<main_metric>
<max_rows_per_second />
<max_bytes_per_second />
<avg_rows_per_second />
<avg_bytes_per_second />
</main_metric>
<main_metric>
<min_time/>
</main_metric>
<substitutions>
<substitution>
@ -38,19 +35,19 @@
<allow_simdjson>1</allow_simdjson>
</settings>
<query>SELECT 'simdjson-1', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam'))</query>
<query>SELECT 'simdjson-2', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'simdjson-3', count() FROM system.numbers WHERE NOT ignore(JSONExtractInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'simdjson-4', count() FROM system.numbers WHERE NOT ignore(JSONExtractUInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'simdjson-5', count() FROM system.numbers WHERE NOT ignore(JSONExtractFloat(materialize({json}), 'fparam'))</query>
<query>SELECT 'simdjson-6', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam'))</query>
<query>SELECT 'simdjson-7', count() FROM system.numbers WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'simdjson-8', count() FROM system.numbers WHERE NOT ignore(JSONExtractInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'simdjson-9', count() FROM system.numbers WHERE NOT ignore(JSONExtractUInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'simdjson-10', count() FROM system.numbers WHERE NOT ignore(JSONExtractRaw(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'simdjson-11', count() FROM system.numbers WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'simdjson-12', count() FROM system.numbers WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam', 'nested_2', -2))</query>
<query>SELECT 'simdjson-13', count() FROM system.numbers WHERE NOT ignore(JSONExtractBool(materialize({long_json}), 'bparam'))</query>
<query>SELECT 'simdjson-1', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam'))</query>
<query>SELECT 'simdjson-2', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'simdjson-3', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'simdjson-4', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractUInt(materialize({json}), 'nparam'))</query>
<query>SELECT 'simdjson-5', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractFloat(materialize({json}), 'fparam'))</query>
<query>SELECT 'simdjson-6', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam'))</query>
<query>SELECT 'simdjson-7', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractString(materialize({long_json}), 'sparam', 'nested_1'))</query>
<query>SELECT 'simdjson-8', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'simdjson-9', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractUInt(materialize({long_json}), 'nparam'))</query>
<query>SELECT 'simdjson-10', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractRaw(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'simdjson-11', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam'))</query>
<query>SELECT 'simdjson-12', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractFloat(materialize({long_json}), 'fparam', 'nested_2', -2))</query>
<query>SELECT 'simdjson-13', count() FROM numbers(1000000) WHERE NOT ignore(JSONExtractBool(materialize({long_json}), 'bparam'))</query>
</test>

47
libs/libdaemon/src/BaseDaemon.cpp
View File

@ -38,7 +38,7 @@
#include <Poco/DirectoryIterator.h>
#include <Common/Exception.h>
#include <IO/WriteBufferFromFile.h>
#include <IO/WriteBufferFromFileDescriptor.h>
#include <IO/WriteBufferFromFileDescriptorDiscardOnFailure.h>
#include <IO/ReadBufferFromFileDescriptor.h>
#include <IO/ReadHelpers.h>
#include <IO/WriteHelpers.h>
@ -92,18 +92,12 @@ static void terminateRequestedSignalHandler(int sig, siginfo_t * info, void * co
}
thread_local bool already_signal_handled = false;
/** Handler for "fault" signals. Send data about fault to separate thread to write into log.
*/
static void faultSignalHandler(int sig, siginfo_t * info, void * context)
{
if (already_signal_handled)
return;
already_signal_handled = true;
char buf[buf_size];
DB::WriteBufferFromFileDescriptor out(signal_pipe.fds_rw[1], buf_size, buf);
DB::WriteBufferFromFileDescriptorDiscardOnFailure out(signal_pipe.fds_rw[1], buf_size, buf);
const ucontext_t signal_context = *reinterpret_cast<ucontext_t *>(context);
const StackTrace stack_trace(signal_context);
@ -116,10 +110,12 @@ static void faultSignalHandler(int sig, siginfo_t * info, void * context)
out.next();
/// The time that is usually enough for separate thread to print info into log.
::sleep(10);
call_default_signal_handler(sig);
if (sig != SIGPROF) /// This signal is used for debugging.
{
/// The time that is usually enough for separate thread to print info into log.
::sleep(10);
call_default_signal_handler(sig);
}
}
@ -192,7 +188,9 @@ public:
DB::readPODBinary(stack_trace, in);
DB::readBinary(thread_num, in);
onFault(sig, info, context, stack_trace, thread_num);
/// This allows to receive more signals if failure happens inside onFault function.
/// Example: segfault while symbolizing stack trace.
std::thread([=] { onFault(sig, info, context, stack_trace, thread_num); }).detach();
}
}
}
@ -207,14 +205,29 @@ private:
LOG_FATAL(log, "(version " << VERSION_STRING << VERSION_OFFICIAL << ") (from thread " << thread_num << ") " << message);
}
void onFault(int sig, siginfo_t & info, ucontext_t & context, const StackTrace & stack_trace, ThreadNumber thread_num) const
void onFault(int sig, const siginfo_t & info, const ucontext_t & context, const StackTrace & stack_trace, ThreadNumber thread_num) const
{
LOG_FATAL(log, "########################################");
LOG_FATAL(log, "(version " << VERSION_STRING << VERSION_OFFICIAL << ") (from thread " << thread_num << ") "
<< "Received signal " << strsignal(sig) << " (" << sig << ")" << ".");
LOG_FATAL(log, signalToErrorMessage(sig, info, context));
LOG_FATAL(log, stack_trace.toString());
if (stack_trace.getSize())
{
/// Write bare stack trace (addresses) just in case if we will fail to print symbolized stack trace.
/// NOTE This still require memory allocations and mutex lock inside logger. BTW we can also print it to stderr using write syscalls.
std::stringstream bare_stacktrace;
bare_stacktrace << "Stack trace:";
for (size_t i = stack_trace.getOffset(); i < stack_trace.getSize(); ++i)
bare_stacktrace << ' ' << stack_trace.getFrames()[i];
LOG_FATAL(log, bare_stacktrace.rdbuf());
}
/// Write symbolized stack trace line by line for better grep-ability.
stack_trace.toStringEveryLine([&](const std::string & s) { LOG_FATAL(log, s); });
}
};
@ -703,7 +716,9 @@ void BaseDaemon::initializeTerminationAndSignalProcessing()
}
};
add_signal_handler({SIGABRT, SIGSEGV, SIGILL, SIGBUS, SIGSYS, SIGFPE, SIGPIPE}, faultSignalHandler);
/// SIGPROF is added for debugging purposes. To output a stack trace of any running thread at anytime.
add_signal_handler({SIGABRT, SIGSEGV, SIGILL, SIGBUS, SIGSYS, SIGFPE, SIGPIPE, SIGPROF}, faultSignalHandler);
add_signal_handler({SIGHUP, SIGUSR1}, closeLogsSignalHandler);
add_signal_handler({SIGINT, SIGQUIT, SIGTERM}, terminateRequestedSignalHandler);