Merge branch 'master' of github.com:yandex/ClickHouse

2024-10-22 16:30:49 +00:00 · 2019-08-19 12:46:46 +03:00 · 2019-08-19 12:46:46 +03:00 · 6462809338
commit 6462809338
parent 920aea6b83 9c991b161e
64 changed files with 879 additions and 479 deletions
--- a/cmake/find_cpuinfo.cmake
+++ b/cmake/find_cpuinfo.cmake
@ -1,8 +1,8 @@
 option(USE_INTERNAL_CPUINFO_LIBRARY "Set to FALSE to use system cpuinfo library instead of bundled" ${NOT_UNBUNDLED})

 # Now we have no contrib/libcpuinfo, use from system.
-if (USE_INTERNAL_CPUINFO_LIBRARY AND NOT EXISTS "${ClickHouse_SOURCE_DIR}/contrib/libcpuinfo/include")
-   #message (WARNING "submodule contrib/libcpuid is missing. to fix try run: \n git submodule update --init --recursive")
+if (NOT EXISTS "${ClickHouse_SOURCE_DIR}/contrib/libcpuinfo/include")
+   #message (WARNING "submodule contrib/libcpuinfo is missing. to fix try run: \n git submodule update --init --recursive")
   set (USE_INTERNAL_CPUINFO_LIBRARY 0)
   set (MISSING_INTERNAL_CPUINFO_LIBRARY 1)
 endif ()
@ -12,7 +12,7 @@ if(NOT USE_INTERNAL_CPUINFO_LIBRARY)
    find_path(CPUINFO_INCLUDE_DIR NAMES cpuinfo.h PATHS ${CPUINFO_INCLUDE_PATHS})
 endif()

-if(CPUID_LIBRARY AND CPUID_INCLUDE_DIR)
+if(CPUINFO_LIBRARY AND CPUINFO_INCLUDE_DIR)
    set(USE_CPUINFO 1)
 elseif(NOT MISSING_INTERNAL_CPUINFO_LIBRARY)
    set(CPUINFO_INCLUDE_DIR ${ClickHouse_SOURCE_DIR}/contrib/libcpuinfo/include)
--- a/contrib/mariadb-connector-c
+++ b/contrib/mariadb-connector-c
@ -1 +1 @@
-Subproject commit d85d0e98999cd9e28ceb66645999b4a9ce85370e
+Subproject commit c6503d3acc85ca1a7f5e7e38b605d7c9410aac1e
--- a/contrib/mariadb-connector-c-cmake/CMakeLists.txt
+++ b/contrib/mariadb-connector-c-cmake/CMakeLists.txt
@ -31,6 +31,7 @@ ${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/ma_stmt_codec.c
 ${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/ma_string.c
 ${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/ma_time.c
 ${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/ma_tls.c
+${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/secure/openssl_crypt.c
 #${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/secure/gnutls.c
 #${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/secure/ma_schannel.c
 #${MARIADB_CLIENT_SOURCE_DIR}/libmariadb/secure/schannel.c
@ -42,6 +43,7 @@ ${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/mariadb_cleartext.c
 ${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/my_auth.c
 ${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/old_password.c
 ${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/sha256_pw.c
+${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/caching_sha2_pw.c
 #${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/sspi_client.c
 #${MARIADB_CLIENT_SOURCE_DIR}/plugins/auth/sspi_errmsg.c
 ${MARIADB_CLIENT_SOURCE_DIR}/plugins/connection/aurora.c
--- a/contrib/mariadb-connector-c-cmake/linux_x86_64/libmariadb/ma_client_plugin.c
+++ b/contrib/mariadb-connector-c-cmake/linux_x86_64/libmariadb/ma_client_plugin.c
@ -76,17 +76,20 @@ struct st_client_plugin_int *plugin_list[MYSQL_CLIENT_MAX_PLUGINS + MARIADB_CLIE
 static pthread_mutex_t LOCK_load_client_plugin;
 #endif

- extern struct st_mysql_client_plugin mysql_native_password_client_plugin;
- extern struct st_mysql_client_plugin mysql_old_password_client_plugin;
- extern struct st_mysql_client_plugin pvio_socket_client_plugin;
+extern struct st_mysql_client_plugin mysql_native_password_client_plugin;
+extern struct st_mysql_client_plugin mysql_old_password_client_plugin;
+extern struct st_mysql_client_plugin pvio_socket_client_plugin;
+extern struct st_mysql_client_plugin sha256_password_client_plugin;
+extern struct st_mysql_client_plugin caching_sha2_password_client_plugin;


 struct st_mysql_client_plugin *mysql_client_builtins[]=
 {
-     (struct st_mysql_client_plugin *)&mysql_native_password_client_plugin,
-   (struct st_mysql_client_plugin *)&mysql_old_password_client_plugin,
-   (struct st_mysql_client_plugin *)&pvio_socket_client_plugin,
-
+  (struct st_mysql_client_plugin *)&mysql_native_password_client_plugin,
+  (struct st_mysql_client_plugin *)&mysql_old_password_client_plugin,
+  (struct st_mysql_client_plugin *)&pvio_socket_client_plugin,
+  (struct st_mysql_client_plugin *)&sha256_password_client_plugin,
+  (struct st_mysql_client_plugin *)&caching_sha2_password_client_plugin,
  0
 };

--- a/contrib/simdjson
+++ b/contrib/simdjson
@ -1 +1 @@
-Subproject commit 9dfab9d9a4c111690a101ea0a7506a2b2f3fa414
+Subproject commit e9be643db5cf1c29a69bc80ee72d220124a9c50e
--- a/dbms/cmake/version.cmake
+++ b/dbms/cmake/version.cmake
@ -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 "")
--- a/dbms/programs/server/MySQLHandler.cpp
+++ b/dbms/programs/server/MySQLHandler.cpp
@ -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)
--- a/dbms/programs/server/MySQLHandler.h
+++ b/dbms/programs/server/MySQLHandler.h
@ -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;
--- a/dbms/programs/server/users.xml
+++ b/dbms/programs/server/users.xml
@ -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>

--- a/dbms/src/Common/ProfileEvents.cpp
+++ b/dbms/src/Common/ProfileEvents.cpp
@ -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
--- a/dbms/src/Common/QueryProfiler.cpp
+++ b/dbms/src/Common/QueryProfiler.cpp
@ -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);
@ -204,11 +173,13 @@ QueryProfilerBase<ProfilerImpl>::~QueryProfilerBase()
 template <typename ProfilerImpl>
 void QueryProfilerBase<ProfilerImpl>::tryCleanup()
 {
+#if USE_INTERNAL_UNWIND_LIBRARY
    if (timer_id != nullptr && timer_delete(timer_id))
        LOG_ERROR(log, "Failed to delete query profiler timer " + errnoToString(ErrorCodes::CANNOT_DELETE_TIMER));

    if (previous_handler != nullptr && sigaction(pause_signal, previous_handler, nullptr))
        LOG_ERROR(log, "Failed to restore signal handler after query profiler " + errnoToString(ErrorCodes::CANNOT_SET_SIGNAL_HANDLER));
+#endif
 }

 template class QueryProfilerBase<QueryProfilerReal>;
--- a/dbms/src/Common/QueryProfiler.h
+++ b/dbms/src/Common/QueryProfiler.h
@ -1,7 +1,7 @@
 #pragma once

 #include <Core/Types.h>
-
+#include <common/config_common.h>
 #include <signal.h>
 #include <time.h>

@ -43,8 +43,10 @@ private:

    Poco::Logger * log;

+#if USE_INTERNAL_UNWIND_LIBRARY
    /// Timer id from timer_create(2)
    timer_t timer_id = nullptr;
+#endif

    /// Pause signal to interrupt threads to get traces
    int pause_signal;
--- a/dbms/src/Common/StackTrace.cpp
+++ b/dbms/src/Common/StackTrace.cpp
@ -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.
--- a/dbms/src/Common/StackTrace.h
+++ b/dbms/src/Common/StackTrace.h
@ -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();

--- a/dbms/src/Common/tests/CMakeLists.txt
+++ b/dbms/src/Common/tests/CMakeLists.txt
@ -36,7 +36,7 @@ target_include_directories (simple_cache PRIVATE ${DBMS_INCLUDE_DIR})
 target_link_libraries (simple_cache PRIVATE common)

 add_executable (compact_array compact_array.cpp)
-target_link_libraries (compact_array PRIVATE clickhouse_common_io ${Boost_FILESYSTEM_LIBRARY})
+target_link_libraries (compact_array PRIVATE clickhouse_common_io stdc++fs)

 add_executable (radix_sort radix_sort.cpp)
 target_link_libraries (radix_sort PRIVATE clickhouse_common_io)
--- a/dbms/src/Compression/tests/gtest_compressionCodec.cpp
+++ b/dbms/src/Compression/tests/gtest_compressionCodec.cpp
@ -27,7 +27,7 @@
 /// For the expansion of gtest macros.
 #if defined(__clang__)
    #pragma clang diagnostic ignored "-Wdeprecated"
-#elif defined (__GNUC__)
+#elif defined (__GNUC__) && __GNUC__ >= 9
    #pragma GCC diagnostic ignored "-Wdeprecated-copy"
 #endif

--- a/dbms/src/Core/Defines.h
+++ b/dbms/src/Core/Defines.h
@ -92,6 +92,7 @@
 #endif

 /// Check for presence of address sanitizer
+#if !defined(ADDRESS_SANITIZER)
 #if defined(__has_feature)
    #if __has_feature(address_sanitizer)
        #define ADDRESS_SANITIZER 1
@ -99,7 +100,9 @@
 #elif defined(__SANITIZE_ADDRESS__)
    #define ADDRESS_SANITIZER 1
 #endif
+#endif

+#if !defined(THREAD_SANITIZER)
 #if defined(__has_feature)
    #if __has_feature(thread_sanitizer)
        #define THREAD_SANITIZER 1
@ -107,7 +110,9 @@
 #elif defined(__SANITIZE_THREAD__)
    #define THREAD_SANITIZER 1
 #endif
+#endif

+#if !defined(MEMORY_SANITIZER)
 #if defined(__has_feature)
    #if __has_feature(memory_sanitizer)
        #define MEMORY_SANITIZER 1
@ -115,6 +120,7 @@
 #elif defined(__MEMORY_SANITIZER__)
    #define MEMORY_SANITIZER 1
 #endif
+#endif

 /// Explicitly allow undefined behaviour for certain functions. Use it as a function attribute.
 /// It is useful in case when compiler cannot see (and exploit) it, but UBSan can.
--- a/dbms/src/Core/MySQLProtocol.h
+++ b/dbms/src/Core/MySQLProtocol.h
@ -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;
+};
+
+}
+
 }
 }
--- a/dbms/src/Functions/CMakeLists.txt
+++ b/dbms/src/Functions/CMakeLists.txt
@ -25,6 +25,7 @@ target_link_libraries(clickhouse_functions
    PRIVATE
        ${ZLIB_LIBRARIES}
        ${Boost_FILESYSTEM_LIBRARY}
+        ${CMAKE_DL_LIBS}
 )

 if (OPENSSL_CRYPTO_LIBRARY)
--- a/dbms/src/Functions/bitBoolMaskAnd.cpp
+++ b/dbms/src/Functions/bitBoolMaskAnd.cpp
@ -4,6 +4,10 @@

 namespace DB
 {
+    namespace ErrorCodes
+    {
+        extern const int BAD_CAST;
+    }

    /// Working with UInt8: last bit = can be true, previous = can be false (Like dbms/src/Storages/MergeTree/BoolMask.h).
    /// This function provides "AND" operation for BoolMasks.
@ -17,6 +21,8 @@ namespace DB
        template <typename Result = ResultType>
        static inline Result apply(A left, B right)
        {
+            if constexpr (!std::is_same_v<A, ResultType> || !std::is_same_v<B, ResultType>)
+                throw DB::Exception("It's a bug! Only UInt8 type is supported by __bitBoolMaskAnd.", ErrorCodes::BAD_CAST);
            return static_cast<ResultType>(
                    ((static_cast<ResultType>(left) & static_cast<ResultType>(right)) & 1)
                    | ((((static_cast<ResultType>(left) >> 1) | (static_cast<ResultType>(right) >> 1)) & 1) << 1));
--- a/dbms/src/Functions/bitBoolMaskOr.cpp
+++ b/dbms/src/Functions/bitBoolMaskOr.cpp
@ -4,6 +4,10 @@

 namespace DB
 {
+    namespace ErrorCodes
+    {
+        extern const int BAD_CAST;
+    }

    /// Working with UInt8: last bit = can be true, previous = can be false (Like dbms/src/Storages/MergeTree/BoolMask.h).
    /// This function provides "OR" operation for BoolMasks.
@ -17,6 +21,8 @@ namespace DB
        template <typename Result = ResultType>
        static inline Result apply(A left, B right)
        {
+            if constexpr (!std::is_same_v<A, ResultType> || !std::is_same_v<B, ResultType>)
+                throw DB::Exception("It's a bug! Only UInt8 type is supported by __bitBoolMaskOr.", ErrorCodes::BAD_CAST);
            return static_cast<ResultType>(
                    ((static_cast<ResultType>(left) | static_cast<ResultType>(right)) & 1)
                    | ((((static_cast<ResultType>(left) >> 1) & (static_cast<ResultType>(right) >> 1)) & 1) << 1));
--- a/dbms/src/Functions/bitSwapLastTwo.cpp
+++ b/dbms/src/Functions/bitSwapLastTwo.cpp
@ -4,6 +4,10 @@

 namespace DB
 {
+    namespace ErrorCodes
+    {
+        extern const int BAD_CAST;
+    }

    /// Working with UInt8: last bit = can be true, previous = can be false (Like dbms/src/Storages/MergeTree/BoolMask.h).
    /// This function provides "NOT" operation for BoolMasks by swapping last two bits ("can be true" <-> "can be false").
@ -14,6 +18,8 @@ namespace DB

        static inline ResultType NO_SANITIZE_UNDEFINED apply(A a)
        {
+            if constexpr (!std::is_same_v<A, ResultType>)
+                throw DB::Exception("It's a bug! Only UInt8 type is supported by __bitSwapLastTwo.", ErrorCodes::BAD_CAST);
            return static_cast<ResultType>(
                    ((static_cast<ResultType>(a) & 1) << 1) | ((static_cast<ResultType>(a) >> 1) & 1));
        }
--- a/dbms/src/Functions/bitWrapperFunc.cpp
+++ b/dbms/src/Functions/bitWrapperFunc.cpp
@ -4,6 +4,10 @@

 namespace DB
 {
+    namespace ErrorCodes
+    {
+        extern const int BAD_CAST;
+    }

    /// Working with UInt8: last bit = can be true, previous = can be false (Like dbms/src/Storages/MergeTree/BoolMask.h).
    /// This function wraps bool atomic functions
@ -15,7 +19,9 @@ namespace DB

        static inline ResultType NO_SANITIZE_UNDEFINED apply(A a)
        {
-            return a == static_cast<UInt8>(0) ? static_cast<ResultType>(0b10) : static_cast<ResultType >(0b1);
+            if constexpr (!std::is_integral_v<A>)
+                throw DB::Exception("It's a bug! Only integer types are supported by __bitWrapperFunc.", ErrorCodes::BAD_CAST);
+            return a == 0 ? static_cast<ResultType>(0b10) : static_cast<ResultType >(0b1);
        }

 #if USE_EMBEDDED_COMPILER
--- a/dbms/src/IO/WriteBufferFromFileDescriptorDiscardOnFailure.cpp
+++ b/dbms/src/IO/WriteBufferFromFileDescriptorDiscardOnFailure.cpp
@ -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;
+    }
+}
+
+}
--- a/dbms/src/IO/WriteBufferFromFileDescriptorDiscardOnFailure.h
+++ b/dbms/src/IO/WriteBufferFromFileDescriptorDiscardOnFailure.h
@ -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 {}
+};
+
+}
--- a/dbms/src/IO/tests/CMakeLists.txt
+++ b/dbms/src/IO/tests/CMakeLists.txt
@ -59,10 +59,10 @@ target_link_libraries (write_int PRIVATE clickhouse_common_io)

 if (OS_LINUX OR OS_FREEBSD)
    add_executable(write_buffer_aio write_buffer_aio.cpp)
-    target_link_libraries (write_buffer_aio PRIVATE clickhouse_common_io ${Boost_FILESYSTEM_LIBRARY})
+    target_link_libraries (write_buffer_aio PRIVATE clickhouse_common_io stdc++fs)

    add_executable(read_buffer_aio read_buffer_aio.cpp)
-    target_link_libraries (read_buffer_aio PRIVATE clickhouse_common_io ${Boost_FILESYSTEM_LIBRARY})
+    target_link_libraries (read_buffer_aio PRIVATE clickhouse_common_io stdc++fs)
 endif ()

 add_executable (zlib_buffers zlib_buffers.cpp)
--- a/dbms/src/Interpreters/AnalyzedJoin.cpp
+++ b/dbms/src/Interpreters/AnalyzedJoin.cpp
@ -1,7 +1,5 @@
 #include <Interpreters/AnalyzedJoin.h>
 #include <Interpreters/DatabaseAndTableWithAlias.h>
-#include <Interpreters/SyntaxAnalyzer.h>
-#include <Interpreters/ExpressionAnalyzer.h>
 #include <Interpreters/InterpreterSelectWithUnionQuery.h>

 #include <Parsers/ASTExpressionList.h>
@ -50,40 +48,6 @@ size_t AnalyzedJoin::rightKeyInclusion(const String & name) const
    return count;
 }

-ExpressionActionsPtr AnalyzedJoin::createJoinedBlockActions(
-    const NamesAndTypesList & columns_added_by_join,
-    const ASTSelectQuery * select_query_with_join,
-    const Context & context) const
-{
-    if (!select_query_with_join)
-        return nullptr;
-
-    const ASTTablesInSelectQueryElement * join = select_query_with_join->join();
-
-    if (!join)
-        return nullptr;
-
-    const auto & join_params = join->table_join->as<ASTTableJoin &>();
-
-    /// Create custom expression list with join keys from right table.
-    auto expression_list = std::make_shared<ASTExpressionList>();
-    ASTs & children = expression_list->children;
-
-    if (join_params.on_expression)
-        for (const auto & join_right_key : key_asts_right)
-            children.emplace_back(join_right_key);
-
-    NameSet required_columns_set(key_names_right.begin(), key_names_right.end());
-    for (const auto & joined_column : columns_added_by_join)
-        required_columns_set.insert(joined_column.name);
-    Names required_columns(required_columns_set.begin(), required_columns_set.end());
-
-    ASTPtr query = expression_list;
-    auto syntax_result = SyntaxAnalyzer(context).analyze(query, columns_from_joined_table, required_columns);
-    ExpressionAnalyzer analyzer(query, syntax_result, context);
-    return analyzer.getActions(true, false);
-}
-
 void AnalyzedJoin::deduplicateAndQualifyColumnNames(const NameSet & left_table_columns, const String & right_table_prefix)
 {
    NameSet joined_columns;
--- a/dbms/src/Interpreters/AnalyzedJoin.h
+++ b/dbms/src/Interpreters/AnalyzedJoin.h
@ -14,9 +14,6 @@ class Context;
 class ASTSelectQuery;
 struct DatabaseAndTableWithAlias;

-class ExpressionActions;
-using ExpressionActionsPtr = std::shared_ptr<ExpressionActions>;
-
 struct AnalyzedJoin
 {
    /** Query of the form `SELECT expr(x) AS k FROM t1 ANY LEFT JOIN (SELECT expr(x) AS k FROM t2) USING k`
@ -56,10 +53,8 @@ public:
    void addUsingKey(const ASTPtr & ast);
    void addOnKeys(ASTPtr & left_table_ast, ASTPtr & right_table_ast);

-    ExpressionActionsPtr createJoinedBlockActions(
-        const NamesAndTypesList & columns_added_by_join, /// Subset of available_joined_columns.
-        const ASTSelectQuery * select_query_with_join,
-        const Context & context) const;
+    bool hasUsing() const { return with_using; }
+    bool hasOn() const { return !with_using; }

    NameSet getQualifiedColumnsSet() const;
    NameSet getOriginalColumnsSet() const;
--- a/dbms/src/Interpreters/AsynchronousMetrics.cpp
+++ b/dbms/src/Interpreters/AsynchronousMetrics.cpp
@ -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());
--- a/dbms/src/Interpreters/Context.cpp
+++ b/dbms/src/Interpreters/Context.cpp
@ -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)
 {
--- a/dbms/src/Interpreters/Context.h
+++ b/dbms/src/Interpreters/Context.h
@ -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();

--- a/dbms/src/Interpreters/ExpressionAnalyzer.cpp
+++ b/dbms/src/Interpreters/ExpressionAnalyzer.cpp
@ -69,7 +69,7 @@ using LogAST = DebugASTLog<false>; /// set to true to enable logs
 namespace ErrorCodes
 {
    extern const int UNKNOWN_IDENTIFIER;
-    extern const int EXPECTED_ALL_OR_ANY;
+    extern const int LOGICAL_ERROR;
 }

 ExpressionAnalyzer::ExpressionAnalyzer(
@ -140,7 +140,7 @@ void ExpressionAnalyzer::analyzeAggregation()
                for (const auto & key_ast : analyzedJoin().key_asts_left)
                    getRootActions(key_ast, true, temp_actions);

-            addJoinAction(temp_actions, true);
+            addJoinAction(temp_actions);
        }
    }

@ -411,17 +411,6 @@ bool SelectQueryExpressionAnalyzer::appendArrayJoin(ExpressionActionsChain & cha
    return true;
 }

-void ExpressionAnalyzer::addJoinAction(ExpressionActionsPtr & actions, bool only_types) const
-{
-    if (only_types)
-        actions->add(ExpressionAction::ordinaryJoin(nullptr, analyzedJoin().key_names_left, columnsAddedByJoin()));
-    else
-        for (auto & subquery_for_set : subqueries_for_sets)
-            if (subquery_for_set.second.join)
-                actions->add(ExpressionAction::ordinaryJoin(subquery_for_set.second.join, analyzedJoin().key_names_left,
-                                                            columnsAddedByJoin()));
-}
-
 static void appendRequiredColumns(
    NameSet & required_columns, const Block & sample, const Names & key_names_right, const NamesAndTypesList & columns_added_by_join)
 {
@ -434,48 +423,37 @@ static void appendRequiredColumns(
            required_columns.insert(column.name);
 }

+/// It's possible to set nullptr as join for only_types mode
+void ExpressionAnalyzer::addJoinAction(ExpressionActionsPtr & actions, JoinPtr join) const
+{
+    actions->add(ExpressionAction::ordinaryJoin(join, analyzedJoin().key_names_left, columnsAddedByJoin()));
+}
+
 bool SelectQueryExpressionAnalyzer::appendJoin(ExpressionActionsChain & chain, bool only_types)
 {
-    const auto * select_query = getSelectQuery();
-
-    if (!select_query->join())
+    const ASTTablesInSelectQueryElement * ast_join = getSelectQuery()->join();
+    if (!ast_join)
        return false;

+    SubqueryForSet & subquery_for_set = getSubqueryForJoin(*ast_join);
+
+    ASTPtr left_keys_list = std::make_shared<ASTExpressionList>();
+    left_keys_list->children = analyzedJoin().key_asts_left;
+
    initChain(chain, sourceColumns());
    ExpressionActionsChain::Step & step = chain.steps.back();

-    const auto & join_element = select_query->join()->as<ASTTablesInSelectQueryElement &>();
+    getRootActions(left_keys_list, only_types, step.actions);
+    addJoinAction(step.actions, subquery_for_set.join);
+
+    return true;
+}
+
+static JoinPtr tryGetStorageJoin(const ASTTablesInSelectQueryElement & join_element, const Context & context)
+{
+    const auto & table_to_join = join_element.table_expression->as<ASTTableExpression &>();
    auto & join_params = join_element.table_join->as<ASTTableJoin &>();

-    if (join_params.strictness == ASTTableJoin::Strictness::Unspecified && join_params.kind != ASTTableJoin::Kind::Cross)
-    {
-        if (settings.join_default_strictness == "ANY")
-            join_params.strictness = ASTTableJoin::Strictness::Any;
-        else if (settings.join_default_strictness == "ALL")
-            join_params.strictness = ASTTableJoin::Strictness::All;
-        else
-            throw Exception("Expected ANY or ALL in JOIN section, because setting (join_default_strictness) is empty", DB::ErrorCodes::EXPECTED_ALL_OR_ANY);
-    }
-
-    if (join_params.using_expression_list)
-    {
-        getRootActions(join_params.using_expression_list, only_types, step.actions);
-    }
-    else if (join_params.on_expression)
-    {
-        auto list = std::make_shared<ASTExpressionList>();
-        list->children = analyzedJoin().key_asts_left;
-        getRootActions(list, only_types, step.actions);
-    }
-
-    const auto & table_to_join = join_element.table_expression->as<ASTTableExpression &>();
-
-    /// Two JOINs are not supported with the same subquery, but different USINGs.
-    auto join_hash = join_element.getTreeHash();
-
-    SubqueryForSet & subquery_for_set = subqueries_for_sets[toString(join_hash.first) + "_" + toString(join_hash.second)];
-
-    /// Special case - if table name is specified on the right of JOIN, then the table has the type Join (the previously prepared mapping).
    /// TODO This syntax does not support specifying a database name.
    if (table_to_join.database_and_table_name)
    {
@ -491,64 +469,100 @@ bool SelectQueryExpressionAnalyzer::appendJoin(ExpressionActionsChain & chain, b
                storage_join->assertCompatible(join_params.kind, join_params.strictness);
                /// TODO Check the set of keys.

-                JoinPtr & join = storage_join->getJoin();
-                subquery_for_set.join = join;
+                return storage_join->getJoin();
            }
        }
    }

+    return {};
+}
+
+SubqueryForSet & SelectQueryExpressionAnalyzer::getSubqueryForJoin(const ASTTablesInSelectQueryElement & join_element)
+{
+    /// Two JOINs are not supported with the same subquery, but different USINGs.
+    auto join_hash = join_element.getTreeHash();
+    String join_subquery_id = toString(join_hash.first) + "_" + toString(join_hash.second);
+
+    SubqueryForSet & subquery_for_set = subqueries_for_sets[join_subquery_id];
+
+    /// Special case - if table name is specified on the right of JOIN, then the table has the type Join (the previously prepared mapping).
    if (!subquery_for_set.join)
+        subquery_for_set.join = tryGetStorageJoin(join_element, context);
+
+    if (!subquery_for_set.join)
+        makeHashJoin(join_element, subquery_for_set);
+
+    return subquery_for_set;
+}
+
+void SelectQueryExpressionAnalyzer::makeHashJoin(const ASTTablesInSelectQueryElement & join_element,
+                                                 SubqueryForSet & subquery_for_set) const
+{
+    /// Actions which need to be calculated on joined block.
+    ExpressionActionsPtr joined_block_actions = createJoinedBlockActions();
+
+    /** For GLOBAL JOINs (in the case, for example, of the push method for executing GLOBAL subqueries), the following occurs
+        * - in the addExternalStorage function, the JOIN (SELECT ...) subquery is replaced with JOIN _data1,
+        *   in the subquery_for_set object this subquery is exposed as source and the temporary table _data1 as the `table`.
+        * - this function shows the expression JOIN _data1.
+        */
+    if (!subquery_for_set.source)
    {
+        ASTPtr table;
+
+        auto & table_to_join = join_element.table_expression->as<ASTTableExpression &>();
+        if (table_to_join.subquery)
+            table = table_to_join.subquery;
+        else if (table_to_join.table_function)
+            table = table_to_join.table_function;
+        else if (table_to_join.database_and_table_name)
+            table = table_to_join.database_and_table_name;
+
+        Names action_columns = joined_block_actions->getRequiredColumns();
+        NameSet required_columns(action_columns.begin(), action_columns.end());
+
        auto & analyzed_join = analyzedJoin();
-        /// Actions which need to be calculated on joined block.
-        ExpressionActionsPtr joined_block_actions =
-            analyzed_join.createJoinedBlockActions(columnsAddedByJoin(), select_query, context);
+        appendRequiredColumns(
+            required_columns, joined_block_actions->getSampleBlock(), analyzed_join.key_names_right, columnsAddedByJoin());

-        /** For GLOBAL JOINs (in the case, for example, of the push method for executing GLOBAL subqueries), the following occurs
-          * - in the addExternalStorage function, the JOIN (SELECT ...) subquery is replaced with JOIN _data1,
-          *   in the subquery_for_set object this subquery is exposed as source and the temporary table _data1 as the `table`.
-          * - this function shows the expression JOIN _data1.
-          */
-        if (!subquery_for_set.source)
-        {
-            ASTPtr table;
+        auto original_map = analyzed_join.getOriginalColumnsMap(required_columns);
+        Names original_columns;
+        for (auto & pr : original_map)
+            original_columns.push_back(pr.second);

-            if (table_to_join.subquery)
-                table = table_to_join.subquery;
-            else if (table_to_join.table_function)
-                table = table_to_join.table_function;
-            else if (table_to_join.database_and_table_name)
-                table = table_to_join.database_and_table_name;
+        auto interpreter = interpretSubquery(table, context, subquery_depth, original_columns);

-            Names action_columns = joined_block_actions->getRequiredColumns();
-            NameSet required_columns(action_columns.begin(), action_columns.end());
-
-            appendRequiredColumns(
-                required_columns, joined_block_actions->getSampleBlock(), analyzed_join.key_names_right, columnsAddedByJoin());
-
-            auto original_map = analyzed_join.getOriginalColumnsMap(required_columns);
-            Names original_columns;
-            for (auto & pr : original_map)
-                original_columns.push_back(pr.second);
-
-            auto interpreter = interpretSubquery(table, context, subquery_depth, original_columns);
-
-            subquery_for_set.makeSource(interpreter, original_map);
-        }
-
-        Block sample_block = subquery_for_set.renamedSampleBlock();
-        joined_block_actions->execute(sample_block);
-
-        /// TODO You do not need to set this up when JOIN is only needed on remote servers.
-        subquery_for_set.join = std::make_shared<Join>(analyzedJoin().key_names_right, settings.join_use_nulls,
-            settings.size_limits_for_join, join_params.kind, join_params.strictness);
-        subquery_for_set.join->setSampleBlock(sample_block);
-        subquery_for_set.joined_block_actions = joined_block_actions;
+        subquery_for_set.makeSource(interpreter, original_map);
    }

-    addJoinAction(step.actions, false);
+    Block sample_block = subquery_for_set.renamedSampleBlock();
+    joined_block_actions->execute(sample_block);

-    return true;
+    /// TODO You do not need to set this up when JOIN is only needed on remote servers.
+    auto & join_params = join_element.table_join->as<ASTTableJoin &>();
+    subquery_for_set.join = std::make_shared<Join>(analyzedJoin().key_names_right, settings.join_use_nulls,
+        settings.size_limits_for_join, join_params.kind, join_params.strictness);
+    subquery_for_set.join->setSampleBlock(sample_block);
+    subquery_for_set.joined_block_actions = joined_block_actions;
+}
+
+ExpressionActionsPtr SelectQueryExpressionAnalyzer::createJoinedBlockActions() const
+{
+    /// Create custom expression list with join keys from right table.
+    ASTPtr expression_list = std::make_shared<ASTExpressionList>();
+    ASTs & children = expression_list->children;
+
+    if (analyzedJoin().hasOn())
+        for (const auto & join_right_key : analyzedJoin().key_asts_right)
+            children.emplace_back(join_right_key);
+
+    NameSet required_columns_set(analyzedJoin().key_names_right.begin(), analyzedJoin().key_names_right.end());
+    for (const auto & joined_column : columnsAddedByJoin())
+        required_columns_set.insert(joined_column.name);
+    Names required_columns(required_columns_set.begin(), required_columns_set.end());
+
+    auto syntax_result = SyntaxAnalyzer(context).analyze(expression_list, analyzedJoin().columns_from_joined_table, required_columns);
+    return ExpressionAnalyzer(expression_list, syntax_result, context).getActions(true, false);
 }

 bool SelectQueryExpressionAnalyzer::appendPrewhere(
--- a/dbms/src/Interpreters/ExpressionAnalyzer.h
+++ b/dbms/src/Interpreters/ExpressionAnalyzer.h
@ -24,6 +24,7 @@ struct ASTTableJoin;
 class ASTFunction;
 class ASTExpressionList;
 class ASTSelectQuery;
+struct ASTTablesInSelectQueryElement;

 struct SyntaxAnalyzerResult;
 using SyntaxAnalyzerResultPtr = std::shared_ptr<const SyntaxAnalyzerResult>;
@ -63,14 +64,12 @@ private:
        const bool join_use_nulls;
        const SizeLimits size_limits_for_set;
        const SizeLimits size_limits_for_join;
-        const String join_default_strictness;

        ExtractedSettings(const Settings & settings_)
        :   use_index_for_in_with_subqueries(settings_.use_index_for_in_with_subqueries),
            join_use_nulls(settings_.join_use_nulls),
            size_limits_for_set(settings_.max_rows_in_set, settings_.max_bytes_in_set, settings_.set_overflow_mode),
-            size_limits_for_join(settings_.max_rows_in_join, settings_.max_bytes_in_join, settings_.join_overflow_mode),
-            join_default_strictness(settings_.join_default_strictness.toString())
+            size_limits_for_join(settings_.max_rows_in_join, settings_.max_bytes_in_join, settings_.join_overflow_mode)
        {}
    };

@ -132,7 +131,7 @@ protected:

    void addMultipleArrayJoinAction(ExpressionActionsPtr & actions, bool is_left) const;

-    void addJoinAction(ExpressionActionsPtr & actions, bool only_types) const;
+    void addJoinAction(ExpressionActionsPtr & actions, JoinPtr join = {}) const;

    void getRootActions(const ASTPtr & ast, bool no_subqueries, ExpressionActionsPtr & actions, bool only_consts = false);

@ -224,6 +223,10 @@ private:
      */
    void tryMakeSetForIndexFromSubquery(const ASTPtr & subquery_or_table_name);

+    SubqueryForSet & getSubqueryForJoin(const ASTTablesInSelectQueryElement & join_element);
+    ExpressionActionsPtr createJoinedBlockActions() const;
+    void makeHashJoin(const ASTTablesInSelectQueryElement & join_element, SubqueryForSet & subquery_for_set) const;
+
    const ASTSelectQuery * getAggregatingQuery() const;
 };

--- a/dbms/src/Interpreters/MetricLog.cpp
+++ b/dbms/src/Interpreters/MetricLog.cpp
@ -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.
--- a/dbms/src/Interpreters/MetricLog.h
+++ b/dbms/src/Interpreters/MetricLog.h
@ -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;
--- a/dbms/src/Interpreters/SyntaxAnalyzer.cpp
+++ b/dbms/src/Interpreters/SyntaxAnalyzer.cpp
@ -1,7 +1,9 @@
+#include <Core/Settings.h>
+#include <Core/NamesAndTypes.h>
+
 #include <Interpreters/SyntaxAnalyzer.h>
 #include <Interpreters/InJoinSubqueriesPreprocessor.h>
 #include <Interpreters/LogicalExpressionsOptimizer.h>
-#include <Core/Settings.h>
 #include <Interpreters/QueryAliasesVisitor.h>
 #include <Interpreters/InterpreterSelectWithUnionQuery.h>
 #include <Interpreters/ArrayJoinedColumnsVisitor.h>
@ -29,10 +31,8 @@

 #include <DataTypes/NestedUtils.h>

-#include <Core/NamesAndTypes.h>
 #include <IO/WriteHelpers.h>
 #include <Storages/IStorage.h>
-#include <Common/typeid_cast.h>

 #include <functional>

@ -48,6 +48,7 @@ namespace ErrorCodes
    extern const int EMPTY_LIST_OF_COLUMNS_QUERIED;
    extern const int NOT_IMPLEMENTED;
    extern const int UNKNOWN_IDENTIFIER;
+    extern const int EXPECTED_ALL_OR_ANY;
 }

 NameSet removeDuplicateColumns(NamesAndTypesList & columns)
@ -487,6 +488,27 @@ void getArrayJoinedColumns(ASTPtr & query, SyntaxAnalyzerResult & result, const
    }
 }

+void setJoinStrictness(ASTSelectQuery & select_query, JoinStrictness join_default_strictness)
+{
+    const ASTTablesInSelectQueryElement * node = select_query.join();
+    if (!node)
+        return;
+
+    auto & table_join = const_cast<ASTTablesInSelectQueryElement *>(node)->table_join->as<ASTTableJoin &>();
+
+    if (table_join.strictness == ASTTableJoin::Strictness::Unspecified &&
+        table_join.kind != ASTTableJoin::Kind::Cross)
+    {
+        if (join_default_strictness == JoinStrictness::ANY)
+            table_join.strictness = ASTTableJoin::Strictness::Any;
+        else if (join_default_strictness == JoinStrictness::ALL)
+            table_join.strictness = ASTTableJoin::Strictness::All;
+        else
+            throw Exception("Expected ANY or ALL in JOIN section, because setting (join_default_strictness) is empty",
+                            DB::ErrorCodes::EXPECTED_ALL_OR_ANY);
+    }
+}
+
 /// Find the columns that are obtained by JOIN.
 void collectJoinedColumns(AnalyzedJoin & analyzed_join, const ASTSelectQuery & select_query, const NameSet & source_columns,
                          const Aliases & aliases, bool join_use_nulls)
@ -863,6 +885,7 @@ SyntaxAnalyzerResultPtr SyntaxAnalyzer::analyze(
        /// Push the predicate expression down to the subqueries.
        result.rewrite_subqueries = PredicateExpressionsOptimizer(select_query, settings, context).optimize();

+        setJoinStrictness(*select_query, settings.join_default_strictness);
        collectJoinedColumns(result.analyzed_join, *select_query, source_columns_set, result.aliases, settings.join_use_nulls);
    }

--- a/dbms/src/Interpreters/Users.cpp
+++ b/dbms/src/Interpreters/Users.cpp
@ -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");

--- a/dbms/src/Interpreters/Users.h
+++ b/dbms/src/Interpreters/Users.h
@ -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;
--- a/dbms/src/Interpreters/UsersManager.cpp
+++ b/dbms/src/Interpreters/UsersManager.cpp
@ -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();
--- a/dbms/src/Interpreters/tests/CMakeLists.txt
+++ b/dbms/src/Interpreters/tests/CMakeLists.txt
@ -56,7 +56,7 @@ target_link_libraries (expression_analyzer PRIVATE dbms clickhouse_storages_syst
 add_check(expression_analyzer)

 add_executable (users users.cpp)
-target_link_libraries (users PRIVATE dbms clickhouse_common_config ${Boost_FILESYSTEM_LIBRARY})
+target_link_libraries (users PRIVATE dbms clickhouse_common_config stdc++fs)

 if (OS_LINUX)
    add_executable (internal_iotop internal_iotop.cpp)
--- a/dbms/src/Processors/Formats/Impl/CapnProtoRowInputFormat.cpp
+++ b/dbms/src/Processors/Formats/Impl/CapnProtoRowInputFormat.cpp
@ -1,14 +1,13 @@
-#include "config_formats.h"
-#include <Processors/Formats/Impl/CapnProtoRowInputFormat.h> // Y_IGNORE
+#include "CapnProtoRowInputFormat.h"
 #if USE_CAPNP

 #include <IO/ReadBuffer.h>
 #include <Interpreters/Context.h>
 #include <Formats/FormatFactory.h>
 #include <Formats/FormatSchemaInfo.h>
-#include <capnp/serialize.h> // Y_IGNORE
-#include <capnp/dynamic.h> // Y_IGNORE
-#include <capnp/common.h> // Y_IGNORE
+#include <capnp/serialize.h>
+#include <capnp/dynamic.h>
+#include <capnp/common.h>
 #include <boost/algorithm/string.hpp>
 #include <boost/range/join.hpp>
 #include <common/logger_useful.h>
--- a/dbms/src/Processors/Formats/Impl/CapnProtoRowInputFormat.h
+++ b/dbms/src/Processors/Formats/Impl/CapnProtoRowInputFormat.h
@ -1,10 +1,10 @@
 #pragma once
-#include <Common/config.h>
+
+#include "config_formats.h"
 #if USE_CAPNP

 #include <Core/Block.h>
 #include <Processors/Formats/IRowInputFormat.h>
-
 #include <capnp/schema-parser.h>

 namespace DB
--- a/dbms/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp
+++ b/dbms/src/Processors/Formats/Impl/ParquetBlockInputFormat.cpp
@ -1,7 +1,6 @@
-#include "config_formats.h"
-#include <Processors/Formats/Impl/ParquetBlockInputFormat.h>
-
+#include "ParquetBlockInputFormat.h"
 #if USE_PARQUET
+
 #include <algorithm>
 #include <iterator>
 #include <vector>
--- a/dbms/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp
+++ b/dbms/src/Processors/Formats/Impl/ParquetBlockOutputFormat.cpp
@ -1,5 +1,4 @@
-#include "config_formats.h"
-#include <Processors/Formats/Impl/ParquetBlockOutputFormat.h>
+#include "ParquetBlockOutputFormat.h"

 #if USE_PARQUET

--- a/dbms/src/Processors/Formats/Impl/ProtobufRowInputFormat.cpp
+++ b/dbms/src/Processors/Formats/Impl/ProtobufRowInputFormat.cpp
@ -1,5 +1,4 @@
-#include "config_formats.h"
-#include <Processors/Formats/Impl/ProtobufRowInputFormat.h>
+#include "ProtobufRowInputFormat.h"

 #if USE_PROTOBUF
 #include <Core/Block.h>
--- a/dbms/src/Processors/Formats/Impl/ProtobufRowInputFormat.h
+++ b/dbms/src/Processors/Formats/Impl/ProtobufRowInputFormat.h
@ -1,6 +1,6 @@
 #pragma once

-#include <Common/config.h>
+#include "config_formats.h"
 #if USE_PROTOBUF

 #include <DataTypes/IDataType.h>
--- a/dbms/src/Processors/Formats/Impl/ProtobufRowOutputFormat.cpp
+++ b/dbms/src/Processors/Formats/Impl/ProtobufRowOutputFormat.cpp
@ -1,10 +1,8 @@
 #include <Formats/FormatFactory.h>
+#include "ProtobufRowOutputFormat.h"

-#include "config_formats.h"
 #if USE_PROTOBUF

-#include <Processors/Formats/Impl/ProtobufRowOutputFormat.h>
-
 #include <Core/Block.h>
 #include <Formats/FormatSchemaInfo.h>
 #include <Formats/ProtobufSchemas.h>
--- a/dbms/src/Processors/Formats/Impl/ProtobufRowOutputFormat.h
+++ b/dbms/src/Processors/Formats/Impl/ProtobufRowOutputFormat.h
@ -1,6 +1,6 @@
 #pragma once

-#include <Common/config.h>
+#include "config_formats.h"
 #if USE_PROTOBUF

 #include <Core/Block.h>
--- a/dbms/src/Storages/MergeTree/MergeTreeDataPart.cpp
+++ b/dbms/src/Storages/MergeTree/MergeTreeDataPart.cpp
@ -409,6 +409,8 @@ void MergeTreeDataPart::remove() const
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wunused-variable"
 #endif
+        std::shared_lock<std::shared_mutex> lock(columns_lock);
+
        for (const auto & [file, _] : checksums.files)
        {
            String path_to_remove = to + "/" + file;
--- a/dbms/src/Storages/System/StorageSystemParts.cpp
+++ b/dbms/src/Storages/System/StorageSystemParts.cpp
@ -114,7 +114,11 @@ void StorageSystemParts::processNextStorage(MutableColumns & columns_, const Sto
            columns_[i++]->insert(part->stateString());

        MinimalisticDataPartChecksums helper;
-        helper.computeTotalChecksums(part->checksums);
+        {
+            /// TODO: MergeTreeDataPart structure is too error-prone.
+            std::shared_lock<std::shared_mutex> lock(part->columns_lock);
+            helper.computeTotalChecksums(part->checksums);
+        }

        auto checksum = helper.hash_of_all_files;
        columns_[i++]->insert(getHexUIntLowercase(checksum.first) + getHexUIntLowercase(checksum.second));
--- a/dbms/tests/integration/test_mysql_protocol/clients/mysqljs/Dockerfile
+++ b/dbms/tests/integration/test_mysql_protocol/clients/mysqljs/Dockerfile
@ -0,0 +1,5 @@
+FROM node:8
+
+RUN npm install mysql
+
+COPY ./test.js test.js
--- a/dbms/tests/integration/test_mysql_protocol/clients/mysqljs/docker_compose.yml
+++ b/dbms/tests/integration/test_mysql_protocol/clients/mysqljs/docker_compose.yml
@ -0,0 +1,8 @@
+version: '2.2'
+services:
+  mysqljs1:
+    build:
+      context: ./
+      network: host
+    # to keep container running
+    command: sleep infinity
--- a/dbms/tests/integration/test_mysql_protocol/clients/mysqljs/test.js
+++ b/dbms/tests/integration/test_mysql_protocol/clients/mysqljs/test.js
@ -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();
--- a/dbms/tests/integration/test_mysql_protocol/configs/users.xml
+++ b/dbms/tests/integration/test_mysql_protocol/configs/users.xml
@ -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>
--- a/dbms/tests/integration/test_mysql_protocol/test.py
+++ b/dbms/tests/integration/test_mysql_protocol/test.py
@ -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
--- a/dbms/tests/performance/json_extract_rapidjson.xml
+++ b/dbms/tests/performance/json_extract_rapidjson.xml
@ -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>
--- a/dbms/tests/performance/json_extract_simdjson.xml
+++ b/dbms/tests/performance/json_extract_simdjson.xml
@ -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>
--- a/dbms/tests/queries/0_stateless/00975_json_hang.sql
+++ b/dbms/tests/queries/0_stateless/00975_json_hang.sql
@ -1 +1 @@
-SELECT DISTINCT JSONExtractRaw(concat('{"x":', rand() % 2 ? 'true' : 'false', '}'), 'x') AS res FROM numbers(100000) ORDER BY res;
+SELECT DISTINCT JSONExtractRaw(concat('{"x":', rand() % 2 ? 'true' : 'false', '}'), 'x') AS res FROM numbers(1000000) ORDER BY res;
--- a/libs/libcommon/CMakeLists.txt
+++ b/libs/libcommon/CMakeLists.txt
@ -119,7 +119,7 @@ target_link_libraries (common
    ${Poco_Foundation_LIBRARY}
    ${CITYHASH_LIBRARIES}
        PRIVATE
-    ${Boost_FILESYSTEM_LIBRARY}
+    stdc++fs
        PUBLIC
    ${Boost_SYSTEM_LIBRARY}
        PRIVATE
--- a/libs/libcommon/include/common/DateLUT.h
+++ b/libs/libcommon/include/common/DateLUT.h
@ -1,6 +1,6 @@
 #pragma once

-#include <common/DateLUTImpl.h>
+#include "DateLUTImpl.h"
 #include <ext/singleton.h>
 #include <unordered_map>
 #include <atomic>
--- a/libs/libcommon/include/common/DateLUTImpl.h
+++ b/libs/libcommon/include/common/DateLUTImpl.h
@ -1,8 +1,8 @@
 #pragma once

-#include <common/Types.h>
-#include <common/DayNum.h>
-#include <common/likely.h>
+#include "Types.h"
+#include "DayNum.h"
+#include "likely.h"
 #include <ctime>
 #include <string>

--- a/libs/libcommon/include/common/likely.h
+++ b/libs/libcommon/include/common/likely.h
@ -1,9 +1,15 @@
-#pragma once
-
 #if defined(_MSC_VER)
-#define likely(x)   (x)
-#define unlikely(x) (x)
+#   if !defined(likely)
+#      define likely(x)   (x)
+#   endif
+#   if !defined(unlikely)
+#      define unlikely(x) (x)
+#   endif
 #else
-#define likely(x)   (__builtin_expect(!!(x), 1))
-#define unlikely(x) (__builtin_expect(!!(x), 0))
+#   if !defined(likely)
+#       define likely(x)   (__builtin_expect(!!(x), 1))
+#   endif
+#   if !defined(unlikely)
+#       define unlikely(x) (__builtin_expect(!!(x), 0))
+#   endif
 #endif
--- a/libs/libcommon/include/common/memory.h
+++ b/libs/libcommon/include/common/memory.h
@ -1,7 +1,7 @@
 #pragma once

 #include <new>
-#include <common/likely.h>
+#include "likely.h"

 #if __has_include(<common/config_common.h>)
 #include <common/config_common.h>
--- a/libs/libdaemon/src/BaseDaemon.cpp
+++ b/libs/libdaemon/src/BaseDaemon.cpp
@ -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);