Merge remote-tracking branch 'remotes/clickhouse/master' into annadevyatova-DOCSUP-4437-nullmod

.gitmodules

@@ -212,3 +212,6 @@
 [submodule "contrib/boringssl"]
 	path = contrib/boringssl
 	url = https://github.com/ClickHouse-Extras/boringssl.git
+[submodule "contrib/NuRaft"]
+	path = contrib/NuRaft
+	url = https://github.com/eBay/NuRaft.git

CMakeLists.txt

@@ -468,6 +468,7 @@ include (cmake/find/rapidjson.cmake)
 include (cmake/find/fastops.cmake)
 include (cmake/find/odbc.cmake)
 include (cmake/find/rocksdb.cmake)
+include (cmake/find/nuraft.cmake)
 
 
 if(NOT USE_INTERNAL_PARQUET_LIBRARY)

cmake/find/nuraft.cmake

+option(ENABLE_NURAFT "Enable NuRaft" ${ENABLE_LIBRARIES})
+
+if (NOT ENABLE_NURAFT)
+    return()
+endif()
+
+if (NOT EXISTS "${ClickHouse_SOURCE_DIR}/contrib/NuRaft/CMakeLists.txt")
+    message (WARNING "submodule contrib/NuRaft is missing. to fix try run: \n git submodule update --init --recursive")
+    message (${RECONFIGURE_MESSAGE_LEVEL} "Can't find internal NuRaft library")
+    set (USE_NURAFT 0)
+    return()
+endif ()
+
+if (NOT OS_FREEBSD)
+    set (USE_NURAFT 1)
+    set (NURAFT_LIBRARY nuraft)
+
+    set (NURAFT_INCLUDE_DIR "${ClickHouse_SOURCE_DIR}/contrib/NuRaft/include")
+
+    message (STATUS "Using NuRaft=${USE_NURAFT}: ${NURAFT_INCLUDE_DIR} : ${NURAFT_LIBRARY}")
+else()
+    set (USE_NURAFT 0)
+    message (STATUS "Using internal NuRaft library on FreeBSD is not supported")
+endif()

contrib/CMakeLists.txt

@@ -307,5 +307,9 @@ if (USE_INTERNAL_ROCKSDB_LIBRARY)
     add_subdirectory(rocksdb-cmake)
 endif()
 
+if (USE_NURAFT)
+  add_subdirectory(nuraft-cmake)
+endif()
+
 add_subdirectory(fast_float)
 

NuRaft @ 410bd149

+Subproject commit 410bd149da84cdde60b4436b02b738749f4e87e1

boost @ 8e259cd2

-Subproject commit 0b98b443aa7bb77d65efd7b23b3b8c8a0ab5f1f3
+Subproject commit 8e259cd2a6b60d75dd17e73432f11bb7b9351bb1

contrib/boost-cmake/CMakeLists.txt

@@ -12,10 +12,12 @@ if (NOT USE_INTERNAL_BOOST_LIBRARY)
         program_options
         regex
         context
+        coroutine
     )
 
     if(Boost_INCLUDE_DIR AND Boost_FILESYSTEM_LIBRARY AND Boost_FILESYSTEM_LIBRARY AND
-       Boost_PROGRAM_OPTIONS_LIBRARY AND Boost_REGEX_LIBRARY AND Boost_SYSTEM_LIBRARY AND Boost_CONTEXT_LIBRARY)
+        Boost_PROGRAM_OPTIONS_LIBRARY AND Boost_REGEX_LIBRARY AND Boost_SYSTEM_LIBRARY AND Boost_CONTEXT_LIBRARY AND
+        Boost_COROUTINE_LIBRARY)
 
         set(EXTERNAL_BOOST_FOUND 1)
 
@@ -29,6 +31,7 @@ if (NOT USE_INTERNAL_BOOST_LIBRARY)
         add_library (_boost_regex INTERFACE)
         add_library (_boost_system INTERFACE)
         add_library (_boost_context INTERFACE)
+        add_library (_boost_coroutine INTERFACE)
 
         target_link_libraries (_boost_filesystem INTERFACE ${Boost_FILESYSTEM_LIBRARY})
         target_link_libraries (_boost_iostreams INTERFACE ${Boost_IOSTREAMS_LIBRARY})
@@ -36,6 +39,7 @@ if (NOT USE_INTERNAL_BOOST_LIBRARY)
         target_link_libraries (_boost_regex INTERFACE ${Boost_REGEX_LIBRARY})
         target_link_libraries (_boost_system INTERFACE ${Boost_SYSTEM_LIBRARY})
         target_link_libraries (_boost_context INTERFACE ${Boost_CONTEXT_LIBRARY})
+        target_link_libraries (_boost_coroutine INTERFACE ${Boost_COROUTINE_LIBRARY})
 
         add_library (boost::filesystem ALIAS _boost_filesystem)
         add_library (boost::iostreams ALIAS _boost_iostreams)
@@ -43,6 +47,7 @@ if (NOT USE_INTERNAL_BOOST_LIBRARY)
         add_library (boost::regex ALIAS _boost_regex)
         add_library (boost::system ALIAS _boost_system)
         add_library (boost::context ALIAS _boost_context)
+        add_library (boost::coroutine ALIAS _boost_coroutine)
     else()
         set(EXTERNAL_BOOST_FOUND 0)
         message (${RECONFIGURE_MESSAGE_LEVEL} "Can't find system boost")
@@ -76,6 +81,10 @@ if (NOT EXTERNAL_BOOST_FOUND)
     add_library (boost::headers_only ALIAS _boost_headers_only)
     target_include_directories (_boost_headers_only SYSTEM BEFORE INTERFACE ${LIBRARY_DIR})
 
+    # asio
+
+    target_compile_definitions (_boost_headers_only INTERFACE BOOST_ASIO_STANDALONE=1)
+
     # iostreams
 
     set (SRCS_IOSTREAMS
@@ -199,4 +208,16 @@ if (NOT EXTERNAL_BOOST_FOUND)
     add_library (_boost_context ${SRCS_CONTEXT})
     add_library (boost::context ALIAS _boost_context)
     target_include_directories (_boost_context PRIVATE ${LIBRARY_DIR})
+
+    # coroutine
+
+    set (SRCS_COROUTINE
+      ${LIBRARY_DIR}/libs/coroutine/detail/coroutine_context.cpp
+      ${LIBRARY_DIR}/libs/coroutine/exceptions.cpp
+      ${LIBRARY_DIR}/libs/coroutine/posix/stack_traits.cpp
+    )
+    add_library (_boost_coroutine ${SRCS_COROUTINE})
+    add_library (boost::coroutine ALIAS _boost_coroutine)
+    target_include_directories (_boost_coroutine PRIVATE ${LIBRARY_DIR})
+    target_link_libraries(_boost_coroutine PRIVATE _boost_context)
 endif ()

contrib/nuraft-cmake/CMakeLists.txt

+set(LIBRARY_DIR ${ClickHouse_SOURCE_DIR}/contrib/NuRaft)
+
+set(SRCS
+    ${LIBRARY_DIR}/src/handle_priority.cxx
+    ${LIBRARY_DIR}/src/buffer_serializer.cxx
+    ${LIBRARY_DIR}/src/peer.cxx
+    ${LIBRARY_DIR}/src/global_mgr.cxx
+    ${LIBRARY_DIR}/src/buffer.cxx
+    ${LIBRARY_DIR}/src/asio_service.cxx
+    ${LIBRARY_DIR}/src/handle_client_request.cxx
+    ${LIBRARY_DIR}/src/raft_server.cxx
+    ${LIBRARY_DIR}/src/snapshot.cxx
+    ${LIBRARY_DIR}/src/handle_commit.cxx
+    ${LIBRARY_DIR}/src/error_code.cxx
+    ${LIBRARY_DIR}/src/crc32.cxx
+    ${LIBRARY_DIR}/src/handle_snapshot_sync.cxx
+    ${LIBRARY_DIR}/src/stat_mgr.cxx
+    ${LIBRARY_DIR}/src/handle_join_leave.cxx
+    ${LIBRARY_DIR}/src/handle_user_cmd.cxx
+    ${LIBRARY_DIR}/src/handle_custom_notification.cxx
+    ${LIBRARY_DIR}/src/handle_vote.cxx
+    ${LIBRARY_DIR}/src/launcher.cxx
+    ${LIBRARY_DIR}/src/srv_config.cxx
+    ${LIBRARY_DIR}/src/snapshot_sync_req.cxx
+    ${LIBRARY_DIR}/src/handle_timeout.cxx
+    ${LIBRARY_DIR}/src/handle_append_entries.cxx
+    ${LIBRARY_DIR}/src/cluster_config.cxx
+)
+
+
+add_library(nuraft ${SRCS})
+
+target_compile_definitions(nuraft PRIVATE USE_BOOST_ASIO=1 BOOST_ASIO_STANDALONE=1)
+
+target_include_directories (nuraft SYSTEM PRIVATE ${LIBRARY_DIR}/include/libnuraft)
+# for some reason include "asio.h" directly without "boost/" prefix.
+target_include_directories (nuraft SYSTEM PRIVATE ${ClickHouse_SOURCE_DIR}/contrib/boost/boost)
+
+target_link_libraries (nuraft PRIVATE boost::headers_only boost::coroutine)
+
+if(OPENSSL_SSL_LIBRARY AND OPENSSL_CRYPTO_LIBRARY)
+    target_link_libraries (nuraft PRIVATE ${OPENSSL_SSL_LIBRARY} ${OPENSSL_CRYPTO_LIBRARY})
+endif()
+
+target_include_directories (nuraft SYSTEM PUBLIC ${LIBRARY_DIR}/include)

docker/packager/unbundled/Dockerfile

@@ -22,6 +22,7 @@ RUN apt-get update \
         libboost-iostreams-dev \
         libboost-regex-dev \
         libboost-context-dev \
+        libboost-coroutine-dev \
         zlib1g-dev \
         liblz4-dev \
         libdouble-conversion-dev \

src/CMakeLists.txt

@@ -307,6 +307,9 @@ if (USE_KRB5)
     dbms_target_link_libraries(PRIVATE ${KRB5_LIBRARY})
 endif()
 
+if (USE_NURAFT)
+    dbms_target_link_libraries(PRIVATE ${NURAFT_LIBRARY})
+endif()
 
 if(RE2_INCLUDE_DIR)
     target_include_directories(clickhouse_common_io SYSTEM BEFORE PUBLIC ${RE2_INCLUDE_DIR})

src/Common/ErrorCodes.cpp

@@ -530,6 +530,7 @@
     M(561, ZSTD_DECODER_FAILED) \
     M(562, TLD_LIST_NOT_FOUND) \
     M(563, CANNOT_READ_MAP_FROM_TEXT) \
+    M(564, INTERSERVER_SCHEME_DOESNT_MATCH) \
     \
     M(999, KEEPER_EXCEPTION) \
     M(1000, POCO_EXCEPTION) \

src/Common/ZooKeeper/TestKeeperStorage.cpp

@@ -740,11 +740,13 @@ void TestKeeperStorage::putRequest(const Coordination::ZooKeeperRequestPtr & req
     request_info.session_id = session_id;
     request_info.response_callback = callback;
 
-    /// Put close requests without timeouts
-    auto timeout = request->getOpNum() == Coordination::OpNum::Close ? 0 : operation_timeout.totalMilliseconds();
     std::lock_guard lock(push_request_mutex);
-    if (!requests_queue.tryPush(std::move(request_info), timeout))
+    /// Put close requests without timeouts
+    if (request->getOpNum() == Coordination::OpNum::Close)
+        requests_queue.push(std::move(request_info));
+    else if (!requests_queue.tryPush(std::move(request_info), operation_timeout.totalMilliseconds()))
         throw Exception("Cannot push request to queue within operation timeout", ErrorCodes::TIMEOUT_EXCEEDED);
+
 }
 
 void TestKeeperStorage::putRequest(const Coordination::ZooKeeperRequestPtr & request, int64_t session_id, ResponseCallback callback, ResponseCallback watch_callback)
@@ -758,10 +760,11 @@ void TestKeeperStorage::putRequest(const Coordination::ZooKeeperRequestPtr & req
     if (request->has_watch)
         request_info.watch_callback = watch_callback;
 
-    /// Put close requests without timeouts
-    auto timeout = request->getOpNum() == Coordination::OpNum::Close ? 0 : operation_timeout.totalMilliseconds();
     std::lock_guard lock(push_request_mutex);
-    if (!requests_queue.tryPush(std::move(request_info), timeout))
+    /// Put close requests without timeouts
+    if (request->getOpNum() == Coordination::OpNum::Close)
+        requests_queue.push(std::move(request_info));
+    else if (!requests_queue.tryPush(std::move(request_info), operation_timeout.totalMilliseconds()))
         throw Exception("Cannot push request to queue within operation timeout", ErrorCodes::TIMEOUT_EXCEEDED);
 }
 

src/Core/DecimalFunctions.h

@@ -6,6 +6,7 @@
 #include <common/arithmeticOverflow.h>
 
 #include <limits>
+#include <type_traits>
 
 
 namespace DB

src/Databases/DatabaseOnDisk.cpp

@@ -103,12 +103,15 @@ String getObjectDefinitionFromCreateQuery(const ASTPtr & query)
         create->attach = true;
 
     /// We remove everything that is not needed for ATTACH from the query.
+    assert(!create->temporary);
     create->database.clear();
     create->as_database.clear();
     create->as_table.clear();
     create->if_not_exists = false;
     create->is_populate = false;
     create->replace_view = false;
+    create->replace_table = false;
+    create->create_or_replace = false;
 
     /// For views it is necessary to save the SELECT query itself, for the rest - on the contrary
     if (!create->is_view && !create->is_materialized_view && !create->is_live_view)

src/Dictionaries/IPAddressDictionary.cpp

@@ -21,10 +21,12 @@ namespace DB
 {
 namespace ErrorCodes
 {
-    extern const int LOGICAL_ERROR;
-    extern const int TYPE_MISMATCH;
     extern const int BAD_ARGUMENTS;
+    extern const int CANNOT_PARSE_INPUT_ASSERTION_FAILED;
+    extern const int CANNOT_PARSE_NUMBER;
     extern const int DICTIONARY_IS_EMPTY;
+    extern const int LOGICAL_ERROR;
+    extern const int TYPE_MISMATCH;
 }
 
 namespace
@@ -89,9 +91,9 @@ static std::pair<Poco::Net::IPAddress, UInt8> parseIPFromString(const std::strin
             const auto * addr_str_end = addr_str.data() + addr_str.size();
             auto [p, ec] = std::from_chars(addr_str.data() + pos + 1, addr_str_end, prefix);
             if (p != addr_str_end)
-                throw DB::Exception("extra characters at the end", ErrorCodes::LOGICAL_ERROR);
+                throw DB::Exception("Extra characters at the end of IP address", ErrorCodes::CANNOT_PARSE_INPUT_ASSERTION_FAILED);
             if (ec != std::errc())
-                throw DB::Exception("mask is not a valid number", ErrorCodes::LOGICAL_ERROR);
+                throw DB::Exception("Mask for IP address is not a valid number", ErrorCodes::CANNOT_PARSE_NUMBER);
 
             addr = addr & Poco::Net::IPAddress(prefix, addr.family());
             return {addr, prefix};
@@ -102,8 +104,8 @@ static std::pair<Poco::Net::IPAddress, UInt8> parseIPFromString(const std::strin
     }
     catch (Poco::Exception & ex)
     {
-        throw DB::Exception("can't parse address \"" + std::string(addr_str) + "\": " + ex.what(),
-            ErrorCodes::LOGICAL_ERROR);
+        throw DB::Exception("Can't parse address \"" + std::string(addr_str) + "\": " + ex.what(),
+            ErrorCodes::CANNOT_PARSE_INPUT_ASSERTION_FAILED);
     }
 }
 

src/Functions/intDiv.cpp

@@ -88,6 +88,8 @@ struct DivideIntegralByConstantImpl
 
 }
 
+namespace impl_
+{
 template <> struct BinaryOperationImpl<UInt64, UInt8, DivideIntegralImpl<UInt64, UInt8>> : DivideIntegralByConstantImpl<UInt64, UInt8> {};
 template <> struct BinaryOperationImpl<UInt64, UInt16, DivideIntegralImpl<UInt64, UInt16>> : DivideIntegralByConstantImpl<UInt64, UInt16> {};
 template <> struct BinaryOperationImpl<UInt64, UInt32, DivideIntegralImpl<UInt64, UInt32>> : DivideIntegralByConstantImpl<UInt64, UInt32> {};
@@ -107,7 +109,7 @@ template <> struct BinaryOperationImpl<Int32, Int8, DivideIntegralImpl<Int32, In
 template <> struct BinaryOperationImpl<Int32, Int16, DivideIntegralImpl<Int32, Int16>> : DivideIntegralByConstantImpl<Int32, Int16> {};
 template <> struct BinaryOperationImpl<Int32, Int32, DivideIntegralImpl<Int32, Int32>> : DivideIntegralByConstantImpl<Int32, Int32> {};
 template <> struct BinaryOperationImpl<Int32, Int64, DivideIntegralImpl<Int32, Int64>> : DivideIntegralByConstantImpl<Int32, Int64> {};
-
+}
 
 struct NameIntDiv { static constexpr auto name = "intDiv"; };
 using FunctionIntDiv = BinaryArithmeticOverloadResolver<DivideIntegralImpl, NameIntDiv, false>;

src/Functions/modulo.cpp

@@ -79,6 +79,8 @@ struct ModuloByConstantImpl
   * Can be expanded to all possible combinations, but more code is needed.
   */
 
+namespace impl_
+{
 template <> struct BinaryOperationImpl<UInt64, UInt8, ModuloImpl<UInt64, UInt8>> : ModuloByConstantImpl<UInt64, UInt8> {};
 template <> struct BinaryOperationImpl<UInt64, UInt16, ModuloImpl<UInt64, UInt16>> : ModuloByConstantImpl<UInt64, UInt16> {};
 template <> struct BinaryOperationImpl<UInt64, UInt32, ModuloImpl<UInt64, UInt32>> : ModuloByConstantImpl<UInt64, UInt32> {};
@@ -98,7 +100,7 @@ template <> struct BinaryOperationImpl<Int32, Int8, ModuloImpl<Int32, Int8>> : M
 template <> struct BinaryOperationImpl<Int32, Int16, ModuloImpl<Int32, Int16>> : ModuloByConstantImpl<Int32, Int16> {};
 template <> struct BinaryOperationImpl<Int32, Int32, ModuloImpl<Int32, Int32>> : ModuloByConstantImpl<Int32, Int32> {};
 template <> struct BinaryOperationImpl<Int32, Int64, ModuloImpl<Int32, Int64>> : ModuloByConstantImpl<Int32, Int64> {};
-
+}
 
 struct NameModulo { static constexpr auto name = "modulo"; };
 using FunctionModulo = BinaryArithmeticOverloadResolver<ModuloImpl, NameModulo, false>;

src/Functions/multiply.cpp

+#include <type_traits>
 #include <Functions/FunctionFactory.h>
 #include <Functions/FunctionBinaryArithmetic.h>
 #include <common/arithmeticOverflow.h>
@@ -25,11 +26,17 @@ struct MultiplyImpl
             return static_cast<Result>(a) * b;
     }
 
-    /// Apply operation and check overflow. It's used for Deciamal operations. @returns true if overflowed, false otherwise.
+    /// Apply operation and check overflow. It's used for Decimal operations. @returns true if overflowed, false otherwise.
     template <typename Result = ResultType>
     static inline bool apply(A a, B b, Result & c)
     {
-        return common::mulOverflow(static_cast<Result>(a), b, c);
+        if constexpr (std::is_same_v<Result, float> || std::is_same_v<Result, double>)
+        {
+            c = static_cast<Result>(a) * b;
+            return false;
+        }
+        else
+            return common::mulOverflow(static_cast<Result>(a), b, c);
     }
 
 #if USE_EMBEDDED_COMPILER

src/Interpreters/InterpreterCreateQuery.cpp

@@ -81,6 +81,7 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
     extern const int PATH_ACCESS_DENIED;
     extern const int NOT_IMPLEMENTED;
+    extern const int UNKNOWN_TABLE;
 }
 
 namespace fs = std::filesystem;
@@ -733,6 +734,36 @@ void InterpreterCreateQuery::assertOrSetUUID(ASTCreateQuery & create, const Data
         /// Ignore UUID if it's ON CLUSTER query
         create.uuid = UUIDHelpers::Nil;
     }
+
+    if (create.replace_table)
+    {
+        if (database->getUUID() == UUIDHelpers::Nil)
+            throw Exception(ErrorCodes::INCORRECT_QUERY,
+                            "{} query is supported only for Atomic databases",
+                            create.create_or_replace ? "CREATE OR REPLACE TABLE" : "REPLACE TABLE");
+
+        UUID uuid_of_table_to_replace;
+        if (create.create_or_replace)
+        {
+            uuid_of_table_to_replace = context.tryResolveStorageID(StorageID(create.database, create.table)).uuid;
+            if (uuid_of_table_to_replace == UUIDHelpers::Nil)
+            {
+                /// Convert to usual CREATE
+                create.replace_table = false;
+                assert(!database->isTableExist(create.table, context));
+            }
+            else
+                create.table = "_tmp_replace_" + toString(uuid_of_table_to_replace);
+        }
+        else
+        {
+            uuid_of_table_to_replace = context.resolveStorageID(StorageID(create.database, create.table)).uuid;
+            if (uuid_of_table_to_replace == UUIDHelpers::Nil)
+                throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {}.{} doesn't exist",
+                                backQuoteIfNeed(create.database), backQuoteIfNeed(create.table));
+            create.table = "_tmp_replace_" + toString(uuid_of_table_to_replace);
+        }
+    }
 }
 
 
@@ -756,9 +787,9 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
         auto query = database->getCreateTableQuery(create.table, context);
         create = query->as<ASTCreateQuery &>(); // Copy the saved create query, but use ATTACH instead of CREATE
         if (create.is_dictionary)
-            throw Exception(
-                "Cannot ATTACH TABLE " + backQuoteIfNeed(database_name) + "." + backQuoteIfNeed(create.table) + ", it is a Dictionary",
-                ErrorCodes::INCORRECT_QUERY);
+            throw Exception(ErrorCodes::INCORRECT_QUERY,
+                            "Cannot ATTACH TABLE {}.{}, it is a Dictionary",
+                            backQuoteIfNeed(database_name), backQuoteIfNeed(create.table));
         create.attach = true;
         create.attach_short_syntax = true;
         create.if_not_exists = if_not_exists;
@@ -804,6 +835,9 @@ BlockIO InterpreterCreateQuery::createTable(ASTCreateQuery & create)
     /// Set and retrieve list of columns, indices and constraints. Set table engine if needed. Rewrite query in canonical way.
     TableProperties properties = setProperties(create);
 
+    if (create.replace_table)
+        return doCreateOrReplaceTable(create, properties);
+
     /// Actually creates table
     bool created = doCreateTable(create, properties);
     if (!created)   /// Table already exists
@@ -820,20 +854,19 @@ bool InterpreterCreateQuery::doCreateTable(ASTCreateQuery & create,
     String data_path;
     DatabasePtr database;
 
-    const String table_name = create.table;
     bool need_add_to_database = !create.temporary;
     if (need_add_to_database)
     {
         /** If the request specifies IF NOT EXISTS, we allow concurrent CREATE queries (which do nothing).
           * If table doesn't exist, one thread is creating table, while others wait in DDLGuard.
           */
-        guard = DatabaseCatalog::instance().getDDLGuard(create.database, table_name);
+        guard = DatabaseCatalog::instance().getDDLGuard(create.database, create.table);
 
         database = DatabaseCatalog::instance().getDatabase(create.database);
         assertOrSetUUID(create, database);
 
         /// Table can be created before or it can be created concurrently in another thread, while we were waiting in DDLGuard.
-        if (database->isTableExist(table_name, context))
+        if (database->isTableExist(create.table, context))
         {
             /// TODO Check structure of table
             if (create.if_not_exists)
@@ -843,27 +876,28 @@ bool InterpreterCreateQuery::doCreateTable(ASTCreateQuery & create,
                 /// when executing CREATE OR REPLACE VIEW, drop current existing view
                 auto drop_ast = std::make_shared<ASTDropQuery>();
                 drop_ast->database = create.database;
-                drop_ast->table = table_name;
+                drop_ast->table = create.table;
                 drop_ast->no_ddl_lock = true;
 
                 InterpreterDropQuery interpreter(drop_ast, context);
                 interpreter.execute();
             }
             else
-                throw Exception("Table " + create.database + "." + table_name + " already exists.", ErrorCodes::TABLE_ALREADY_EXISTS);
+                throw Exception(ErrorCodes::TABLE_ALREADY_EXISTS, "Table {}.{} already exists.", backQuoteIfNeed(create.database), backQuoteIfNeed(create.table));
         }
 
         data_path = database->getTableDataPath(create);
         if (!create.attach && !data_path.empty() && fs::exists(fs::path{context.getPath()} / data_path))
-            throw Exception("Directory for table data " + data_path + " already exists", ErrorCodes::TABLE_ALREADY_EXISTS);
+            throw Exception(ErrorCodes::TABLE_ALREADY_EXISTS, "Directory for table data {} already exists", String(data_path));
     }
     else
     {
-        if (create.if_not_exists && context.tryResolveStorageID({"", table_name}, Context::ResolveExternal))
+        if (create.if_not_exists && context.tryResolveStorageID({"", create.table}, Context::ResolveExternal))
             return false;
 
+        String temporary_table_name = create.table;
         auto temporary_table = TemporaryTableHolder(context, properties.columns, properties.constraints, query_ptr);
-        context.getSessionContext().addExternalTable(table_name, std::move(temporary_table));
+        context.getSessionContext().addExternalTable(temporary_table_name, std::move(temporary_table));
         return true;
     }
 
@@ -903,7 +937,7 @@ bool InterpreterCreateQuery::doCreateTable(ASTCreateQuery & create,
                         "ATTACH ... FROM ... query is not supported for {} table engine, "
                         "because such tables do not store any data on disk. Use CREATE instead.", res->getName());
 
-    database->createTable(context, table_name, res, query_ptr);
+    database->createTable(context, create.table, res, query_ptr);
 
     /// Move table data to the proper place. Wo do not move data earlier to avoid situations
     /// when data directory moved, but table has not been created due to some error.
@@ -927,6 +961,50 @@ bool InterpreterCreateQuery::doCreateTable(ASTCreateQuery & create,
     return true;
 }
 
+
+BlockIO InterpreterCreateQuery::doCreateOrReplaceTable(ASTCreateQuery & create,
+                                                       const InterpreterCreateQuery::TableProperties & properties)
+{
+    auto ast_drop = std::make_shared<ASTDropQuery>();
+    String table_to_replace_name = create.table;
+    bool created = false;
+    bool replaced = false;
+
+    try
+    {
+        [[maybe_unused]] bool done = doCreateTable(create, properties);
+        assert(done);
+        ast_drop->table = create.table;
+        ast_drop->database = create.database;
+        ast_drop->kind = ASTDropQuery::Drop;
+        created = true;
+        if (!create.replace_table)
+            return fillTableIfNeeded(create);
+
+        auto ast_rename = std::make_shared<ASTRenameQuery>();
+        ASTRenameQuery::Element elem
+        {
+            ASTRenameQuery::Table{create.database, create.table},
+            ASTRenameQuery::Table{create.database, table_to_replace_name}
+        };
+        ast_rename->elements.push_back(std::move(elem));
+        ast_rename->exchange = true;
+        InterpreterRenameQuery(ast_rename, context).execute();
+        replaced = true;
+
+        InterpreterDropQuery(ast_drop, context).execute();
+
+        create.table = table_to_replace_name;
+        return fillTableIfNeeded(create);
+    }
+    catch (...)
+    {
+        if (created && create.replace_table && !replaced)
+            InterpreterDropQuery(ast_drop, context).execute();
+        throw;
+    }
+}
+
 BlockIO InterpreterCreateQuery::fillTableIfNeeded(const ASTCreateQuery & create)
 {
     /// If the query is a CREATE SELECT, insert the data into the table.
@@ -1079,22 +1157,22 @@ AccessRightsElements InterpreterCreateQuery::getRequiredAccess() const
     }
     else if (create.is_view || create.is_materialized_view || create.is_live_view)
     {
-        if (create.temporary)
-            required_access.emplace_back(AccessType::CREATE_TEMPORARY_TABLE);
+        assert(!create.temporary);
+        if (create.replace_view)
+            required_access.emplace_back(AccessType::DROP_VIEW | AccessType::CREATE_VIEW, create.database, create.table);
         else
-        {
-            if (create.replace_view)
-                required_access.emplace_back(AccessType::DROP_VIEW | AccessType::CREATE_VIEW, create.database, create.table);
-            else
-                required_access.emplace_back(AccessType::CREATE_VIEW, create.database, create.table);
-        }
+            required_access.emplace_back(AccessType::CREATE_VIEW, create.database, create.table);
     }
     else
     {
         if (create.temporary)
             required_access.emplace_back(AccessType::CREATE_TEMPORARY_TABLE);
         else
+        {
+            if (create.replace_table)
+                required_access.emplace_back(AccessType::DROP_TABLE, create.database, create.table);
             required_access.emplace_back(AccessType::CREATE_TABLE, create.database, create.table);
+        }
     }
 
     if (create.to_table_id)

src/Interpreters/InterpreterCreateQuery.h

@@ -81,6 +81,7 @@ private:
 
     /// Create IStorage and add it to database. If table already exists and IF NOT EXISTS specified, do nothing and return false.
     bool doCreateTable(ASTCreateQuery & create, const TableProperties & properties);
+    BlockIO doCreateOrReplaceTable(ASTCreateQuery & create, const InterpreterCreateQuery::TableProperties & properties);
     /// Inserts data in created table if it's CREATE ... SELECT
     BlockIO fillTableIfNeeded(const ASTCreateQuery & create);
 

src/Parsers/ASTCreateQuery.cpp

@@ -230,19 +230,28 @@ void ASTCreateQuery::formatQueryImpl(const FormatSettings & settings, FormatStat
 
     if (!is_dictionary)
     {
-        std::string what = "TABLE";
+        String action = "CREATE";
+        if (attach)
+            action = "ATTACH";
+        else if (replace_view)
+            action = "CREATE OR REPLACE";
+        else if (replace_table && create_or_replace)
+            action = "CREATE OR REPLACE";
+        else if (replace_table)
+            action = "REPLACE";
+
+        String what = "TABLE";
         if (is_view)
             what = "VIEW";
-        if (is_materialized_view)
+        else if (is_materialized_view)
             what = "MATERIALIZED VIEW";
-        if (is_live_view)
+        else if (is_live_view)
             what = "LIVE VIEW";
 
         settings.ostr
             << (settings.hilite ? hilite_keyword : "")
-                << (attach ? "ATTACH " : "CREATE ")
+                << action << " "
                 << (temporary ? "TEMPORARY " : "")
-                << (replace_view ? "OR REPLACE " : "")
                 << what << " "
                 << (if_not_exists ? "IF NOT EXISTS " : "")
             << (settings.hilite ? hilite_none : "")

src/Parsers/ASTCreateQuery.h

@@ -64,7 +64,7 @@ public:
     bool replace_view{false}; /// CREATE OR REPLACE VIEW
     ASTColumns * columns_list = nullptr;
     ASTExpressionList * tables = nullptr;
-    //FIXME
+
     StorageID to_table_id = StorageID::createEmpty();   /// For CREATE MATERIALIZED VIEW mv TO table.
     ASTStorage * storage = nullptr;
     String as_database;
@@ -81,6 +81,9 @@ public:
 
     std::optional<String> attach_from_path = std::nullopt;
 
+    bool replace_table{false};
+    bool create_or_replace{false};
+
     /** Get the text that identifies this element. */
     String getID(char delim) const override { return (attach ? "AttachQuery" : "CreateQuery") + (delim + database) + delim + table; }
 

src/Parsers/ParserCreateQuery.cpp

@@ -355,6 +355,8 @@ bool ParserCreateTableQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expe
 {
     ParserKeyword s_create("CREATE");
     ParserKeyword s_attach("ATTACH");
+    ParserKeyword s_replace("REPLACE");
+    ParserKeyword s_or_replace("OR REPLACE");
     ParserKeyword s_temporary("TEMPORARY");
     ParserKeyword s_table("TABLE");
     ParserKeyword s_if_not_exists("IF NOT EXISTS");
@@ -383,25 +385,32 @@ bool ParserCreateTableQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expe
 
     String cluster_str;
     bool attach = false;
+    bool replace = false;
+    bool or_replace = false;
     bool if_not_exists = false;
     bool is_temporary = false;
 
-    if (!s_create.ignore(pos, expected))
+    if (s_create.ignore(pos, expected))
     {
-        if (s_attach.ignore(pos, expected))
-            attach = true;
-        else
-            return false;
+        if (s_or_replace.ignore(pos, expected))
+            replace = or_replace = true;
     }
+    else if (s_attach.ignore(pos, expected))
+        attach = true;
+    else if (s_replace.ignore(pos, expected))
+        replace = true;
+    else
+        return false;
 
-    if (s_temporary.ignore(pos, expected))
+
+    if (!replace && !or_replace && s_temporary.ignore(pos, expected))
     {
         is_temporary = true;
     }
     if (!s_table.ignore(pos, expected))
         return false;
 
-    if (s_if_not_exists.ignore(pos, expected))
+    if (!replace && !or_replace && s_if_not_exists.ignore(pos, expected))
         if_not_exists = true;
 
     if (!table_name_p.parse(pos, table, expected))
@@ -505,6 +514,8 @@ bool ParserCreateTableQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expe
         query->as_table_function = as_table_function;
 
     query->attach = attach;
+    query->replace_table = replace;
+    query->create_or_replace = or_replace;
     query->if_not_exists = if_not_exists;
     query->temporary = is_temporary;
 

src/Storages/StorageReplicatedMergeTree.cpp

@@ -122,6 +122,7 @@ namespace ErrorCodes
     extern const int ILLEGAL_TYPE_OF_ARGUMENT;
     extern const int UNKNOWN_POLICY;
     extern const int NO_SUCH_DATA_PART;
+    extern const int INTERSERVER_SCHEME_DOESNT_MATCH;
 }
 
 namespace ActionLocks
@@ -3509,7 +3510,7 @@ bool StorageReplicatedMergeTree::fetchPart(const String & part_name, const Stora
             if (interserver_scheme != address.scheme)
                 throw Exception("Interserver schemes are different: '" + interserver_scheme
                     + "' != '" + address.scheme + "', can't fetch part from " + address.host,
-                    ErrorCodes::LOGICAL_ERROR);
+                    ErrorCodes::INTERSERVER_SCHEME_DOESNT_MATCH);
 
             return fetcher.fetchPart(
                 metadata_snapshot, part_name, source_replica_path,

tests/queries/0_stateless/01185_create_or_replace_table.reference

+t1
+CREATE TABLE test_01185.t1\n(\n    `n` UInt64,\n    `s` String\n)\nENGINE = MergeTree\nORDER BY n\nSETTINGS index_granularity = 8192
+t1
+CREATE TABLE test_01185.t1\n(\n    `n` UInt64,\n    `s` Nullable(String)\n)\nENGINE = MergeTree\nORDER BY n\nSETTINGS index_granularity = 8192
+2	\N
+t1
+CREATE TABLE test_01185.t1\n(\n    `n` UInt64\n)\nENGINE = MergeTree\nORDER BY n\nSETTINGS index_granularity = 8192
+3

tests/queries/0_stateless/01185_create_or_replace_table.sql

+drop database if exists test_01185;
+create database test_01185 engine=Atomic;
+
+replace table test_01185.t1 (n UInt64, s String) engine=MergeTree order by n; -- { serverError 60 }
+show tables from test_01185;
+create or replace table test_01185.t1 (n UInt64, s String) engine=MergeTree order by n;
+show tables from test_01185;
+show create table test_01185.t1;
+
+insert into test_01185.t1 values (1, 'test');
+create or replace table test_01185.t1 (n UInt64, s Nullable(String)) engine=MergeTree order by n;
+insert into test_01185.t1 values (2, null);
+show tables from test_01185;
+show create table test_01185.t1;
+select * from test_01185.t1;
+
+replace table test_01185.t1 (n UInt64) engine=MergeTree order by n;
+insert into test_01185.t1 values (3);
+show tables from test_01185;
+show create table test_01185.t1;
+select * from test_01185.t1;
+
+drop database test_01185;

tests/queries/0_stateless/01603_decimal_mult_float.reference

+2.4
+10.165
+0.00012000000000000002
+150.16500000000002
+7.775900000000001
+56.62269
+598.8376688440277
+299.41883723437786
+0.7485470860550345
+2.245641373854596
+1.641386318314034
+1.641386318314034
+1.641386334333447
+1.641386334333447

tests/queries/0_stateless/01603_decimal_mult_float.sql

+SELECT toDecimal32(2, 2) * 1.2;
+SELECT toDecimal64(0.5, 2) * 20.33;
+SELECT 0.00001 * toDecimal32(12, 2);
+SELECT 30.033 * toDecimal32(5, 1);
+
+CREATE TABLE IF NOT EXISTS test01603 (
+    f64 Float64,
+    d Decimal64(3) DEFAULT toDecimal32(f64, 3),
+    f32 Float32 DEFAULT f64
+) ENGINE=MergeTree() ORDER BY f32;
+
+INSERT INTO test01603(f64) SELECT 1 / (number + 1) FROM system.numbers LIMIT 1000;
+
+SELECT sum(d * 1.1) FROM test01603;
+SELECT sum(8.01 * d) FROM test01603;
+
+SELECT sum(f64 * toDecimal64(80, 2)) FROM test01603;
+SELECT sum(toDecimal64(40, 2) * f32) FROM test01603;
+SELECT sum(f64 * toDecimal64(0.1, 2)) FROM test01603;
+SELECT sum(toDecimal64(0.3, 2) * f32) FROM test01603;
+
+SELECT sum(f64 * d) FROM test01603;
+SELECT sum(d * f64) FROM test01603;
+SELECT sum(f32 * d) FROM test01603;
+SELECT sum(d * f32) FROM test01603;
+
+DROP TABLE IF EXISTS test01603;