Merge branch 'master' into upd_aws_sdk

2024-11-25 17:12:03 +00:00 · 2024-01-29 22:13:17 +01:00 · 2024-01-29 22:13:17 +01:00 · 0677f1f439
commit 0677f1f439
parent 368a26a2aa 69a1935a3e
1096 changed files with 18309 additions and 5909 deletions
--- a/base/poco/Foundation/include/Poco/Logger.h
+++ b/base/poco/Foundation/include/Poco/Logger.h
@ -33,7 +33,8 @@ namespace Poco


 class Exception;
-
+class Logger;
+using LoggerPtr = std::shared_ptr<Logger>;

 class Foundation_API Logger : public Channel
 /// Logger is a special Channel that acts as the main
@ -870,6 +871,11 @@ public:
    /// If the Logger does not yet exist, it is created, based
    /// on its parent logger.

+    static LoggerPtr getShared(const std::string & name);
+    /// Returns a shared pointer to the Logger with the given name.
+    /// If the Logger does not yet exist, it is created, based
+    /// on its parent logger.
+
    static Logger & unsafeGet(const std::string & name);
    /// Returns a reference to the Logger with the given name.
    /// If the Logger does not yet exist, it is created, based
@ -885,6 +891,11 @@ public:
    /// given name. The Logger's Channel and log level as set as
    /// specified.

+    static LoggerPtr createShared(const std::string & name, Channel * pChannel, int level = Message::PRIO_INFORMATION);
+    /// Creates and returns a shared pointer to a Logger with the
+    /// given name. The Logger's Channel and log level as set as
+    /// specified.
+
    static Logger & root();
    /// Returns a reference to the root logger, which is the ultimate
    /// ancestor of all Loggers.
@ -893,13 +904,6 @@ public:
    /// Returns a pointer to the Logger with the given name if it
    /// exists, or a null pointer otherwise.

-    static void destroy(const std::string & name);
-    /// Destroys the logger with the specified name. Does nothing
-    /// if the logger is not found.
-    ///
-    /// After a logger has been destroyed, all references to it
-    /// become invalid.
-
    static void shutdown();
    /// Shuts down the logging framework and releases all
    /// Loggers.
@ -929,8 +933,6 @@ public:
    static const std::string ROOT; /// The name of the root logger ("").

 protected:
-    typedef std::map<std::string, Logger *> LoggerMap;
-
    Logger(const std::string & name, Channel * pChannel, int level);
    ~Logger();

@ -938,6 +940,7 @@ protected:
    void log(const std::string & text, Message::Priority prio, const char * file, int line);

    static std::string format(const std::string & fmt, int argc, std::string argv[]);
+    static Logger & unsafeCreate(const std::string & name, Channel * pChannel, int level = Message::PRIO_INFORMATION);
    static Logger & parent(const std::string & name);
    static void add(Logger * pLogger);
    static Logger * find(const std::string & name);
@ -950,9 +953,6 @@ private:
    std::string _name;
    Channel * _pChannel;
    std::atomic_int _level;
-
-    static LoggerMap * _pLoggerMap;
-    static Mutex _mapMtx;
 };


--- a/base/poco/Foundation/include/Poco/RefCountedObject.h
+++ b/base/poco/Foundation/include/Poco/RefCountedObject.h
@ -38,15 +38,15 @@ public:
    /// Creates the RefCountedObject.
    /// The initial reference count is one.

-    void duplicate() const;
-    /// Increments the object's reference count.
+    size_t duplicate() const;
+    /// Increments the object's reference count, returns reference count before call.

-    void release() const throw();
+    size_t release() const throw();
    /// Decrements the object's reference count
    /// and deletes the object if the count
-    /// reaches zero.
+    /// reaches zero, returns reference count before call.

-    int referenceCount() const;
+    size_t referenceCount() const;
    /// Returns the reference count.

 protected:
@ -57,36 +57,40 @@ private:
    RefCountedObject(const RefCountedObject &);
    RefCountedObject & operator=(const RefCountedObject &);

-    mutable AtomicCounter _counter;
+    mutable std::atomic<size_t> _counter;
 };


 //
 // inlines
 //
-inline int RefCountedObject::referenceCount() const
+inline size_t RefCountedObject::referenceCount() const
 {
-    return _counter.value();
+    return _counter.load(std::memory_order_acquire);
 }


-inline void RefCountedObject::duplicate() const
+inline size_t RefCountedObject::duplicate() const
 {
-    ++_counter;
+    return _counter.fetch_add(1, std::memory_order_acq_rel);
 }


-inline void RefCountedObject::release() const throw()
+inline size_t RefCountedObject::release() const throw()
 {
+    size_t reference_count_before = _counter.fetch_sub(1, std::memory_order_acq_rel);
+
    try
    {
-        if (--_counter == 0)
+        if (reference_count_before == 1)
            delete this;
    }
    catch (...)
    {
        poco_unexpected();
    }
+
+    return reference_count_before;
 }


--- a/base/poco/Foundation/src/Logger.cpp
+++ b/base/poco/Foundation/src/Logger.cpp
@ -20,12 +20,38 @@
 #include "Poco/NumberParser.h"
 #include "Poco/String.h"

+#include <cassert>
+#include <mutex>
+
+namespace
+{
+
+std::mutex & getLoggerMutex()
+{
+	auto get_logger_mutex_placeholder_memory = []()
+	{
+		static char buffer[sizeof(std::mutex)]{};
+		return buffer;
+	};
+
+	static std::mutex * logger_mutex = new (get_logger_mutex_placeholder_memory()) std::mutex();
+	return *logger_mutex;
+}
+
+struct LoggerEntry
+{
+	Poco::Logger * logger;
+	bool owned_by_shared_ptr = false;
+};
+
+using LoggerMap = std::unordered_map<std::string, LoggerEntry>;
+LoggerMap * _pLoggerMap = nullptr;
+
+}

 namespace Poco {


-Logger::LoggerMap* Logger::_pLoggerMap = 0;
-Mutex Logger::_mapMtx;
 const std::string Logger::ROOT;


@ -73,7 +99,7 @@ void Logger::setProperty(const std::string& name, const std::string& value)
 		setChannel(LoggingRegistry::defaultRegistry().channelForName(value));
 	else if (name == "level")
 		setLevel(value);
-	else 
+	else
 		Channel::setProperty(name, value);
 }

@ -112,17 +138,17 @@ void Logger::dump(const std::string& msg, const void* buffer, std::size_t length

 void Logger::setLevel(const std::string& name, int level)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	if (_pLoggerMap)
 	{
 		std::string::size_type len = name.length();
-		for (LoggerMap::iterator it = _pLoggerMap->begin(); it != _pLoggerMap->end(); ++it)
+		for (auto & it : *_pLoggerMap)
 		{
-			if (len == 0 || 
-				(it->first.compare(0, len, name) == 0 && (it->first.length() == len || it->first[len] == '.')))
+			if (len == 0 ||
+				(it.first.compare(0, len, name) == 0 && (it.first.length() == len || it.first[len] == '.')))
 			{
-				it->second->setLevel(level);
+				it.second.logger->setLevel(level);
 			}
 		}
 	}
@ -131,17 +157,17 @@ void Logger::setLevel(const std::string& name, int level)

 void Logger::setChannel(const std::string& name, Channel* pChannel)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	if (_pLoggerMap)
 	{
 		std::string::size_type len = name.length();
-		for (LoggerMap::iterator it = _pLoggerMap->begin(); it != _pLoggerMap->end(); ++it)
+		for (auto & it : *_pLoggerMap)
 		{
 			if (len == 0 ||
-				(it->first.compare(0, len, name) == 0 && (it->first.length() == len || it->first[len] == '.')))
+				(it.first.compare(0, len, name) == 0 && (it.first.length() == len || it.first[len] == '.')))
 			{
-				it->second->setChannel(pChannel);
+				it.second.logger->setChannel(pChannel);
 			}
 		}
 	}
@ -150,17 +176,17 @@ void Logger::setChannel(const std::string& name, Channel* pChannel)

 void Logger::setProperty(const std::string& loggerName, const std::string& propertyName, const std::string& value)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	if (_pLoggerMap)
 	{
 		std::string::size_type len = loggerName.length();
-		for (LoggerMap::iterator it = _pLoggerMap->begin(); it != _pLoggerMap->end(); ++it)
+		for (auto & it : *_pLoggerMap)
 		{
 			if (len == 0 ||
-				(it->first.compare(0, len, loggerName) == 0 && (it->first.length() == len || it->first[len] == '.')))
+				(it.first.compare(0, len, loggerName) == 0 && (it.first.length() == len || it.first[len] == '.')))
 			{
-				it->second->setProperty(propertyName, value);
+				it.second.logger->setProperty(propertyName, value);
 			}
 		}
 	}
@ -280,11 +306,88 @@ void Logger::formatDump(std::string& message, const void* buffer, std::size_t le
 }


+namespace
+{
+
+struct LoggerDeleter
+{
+	void operator()(Poco::Logger * logger)
+	{
+		std::lock_guard<std::mutex> lock(getLoggerMutex());
+
+		/// If logger infrastructure is destroyed just decrement logger reference count
+		if (!_pLoggerMap)
+		{
+			logger->release();
+			return;
+		}
+
+		auto it = _pLoggerMap->find(logger->name());
+		assert(it != _pLoggerMap->end());
+
+		/** If reference count is 1, this means this shared pointer owns logger
+		  * and need destroy it.
+		  */
+		size_t reference_count_before_release = logger->release();
+		if (reference_count_before_release == 1)
+		{
+			assert(it->second.owned_by_shared_ptr);
+			_pLoggerMap->erase(it);
+		}
+	}
+};
+
+
+inline LoggerPtr makeLoggerPtr(Logger & logger)
+{
+	return std::shared_ptr<Logger>(&logger, LoggerDeleter());
+}
+
+}
+
+
 Logger& Logger::get(const std::string& name)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

-	return unsafeGet(name);
+	Logger & logger = unsafeGet(name);
+
+	/** If there are already shared pointer created for this logger
+	  * we need to increment Logger reference count and now logger
+	  * is owned by logger infrastructure.
+	  */
+	auto it = _pLoggerMap->find(name);
+	if (it->second.owned_by_shared_ptr)
+	{
+		it->second.logger->duplicate();
+		it->second.owned_by_shared_ptr = false;
+	}
+
+	return logger;
+}
+
+
+LoggerPtr Logger::getShared(const std::string & name)
+{
+	std::lock_guard<std::mutex> lock(getLoggerMutex());
+	bool logger_exists = _pLoggerMap && _pLoggerMap->contains(name);
+
+	Logger & logger = unsafeGet(name);
+
+	/** If logger already exists, then this shared pointer does not own it.
+	  * If logger does not exists, logger infrastructure could be already destroyed
+	  * or logger was created.
+	  */
+	if (logger_exists)
+	{
+		logger.duplicate();
+	}
+	else if (_pLoggerMap)
+	{
+		_pLoggerMap->find(name)->second.owned_by_shared_ptr = true;
+	}
+
+	return makeLoggerPtr(logger);
 }


@ -310,18 +413,24 @@ Logger& Logger::unsafeGet(const std::string& name)

 Logger& Logger::create(const std::string& name, Channel* pChannel, int level)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

-	if (find(name)) throw ExistsException();
-	Logger* pLogger = new Logger(name, pChannel, level);
-	add(pLogger);
-	return *pLogger;
+	return unsafeCreate(name, pChannel, level);
 }

+LoggerPtr Logger::createShared(const std::string & name, Channel * pChannel, int level)
+{
+	std::lock_guard<std::mutex> lock(getLoggerMutex());
+
+	Logger & logger = unsafeCreate(name, pChannel, level);
+	_pLoggerMap->find(name)->second.owned_by_shared_ptr = true;
+
+	return makeLoggerPtr(logger);
+}

 Logger& Logger::root()
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	return unsafeGet(ROOT);
 }
@ -329,7 +438,7 @@ Logger& Logger::root()

 Logger* Logger::has(const std::string& name)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	return find(name);
 }
@ -337,14 +446,18 @@ Logger* Logger::has(const std::string& name)

 void Logger::shutdown()
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	if (_pLoggerMap)
 	{
-		for (LoggerMap::iterator it = _pLoggerMap->begin(); it != _pLoggerMap->end(); ++it)
+		for (auto & it : *_pLoggerMap)
 		{
-			it->second->release();
+			if (it.second.owned_by_shared_ptr)
+				continue;
+
+			it.second.logger->release();
 		}
+
 		delete _pLoggerMap;
 		_pLoggerMap = 0;
 	}
@ -357,31 +470,15 @@ Logger* Logger::find(const std::string& name)
 	{
 		LoggerMap::iterator it = _pLoggerMap->find(name);
 		if (it != _pLoggerMap->end())
-			return it->second;
+			return it->second.logger;
 	}
 	return 0;
 }


-void Logger::destroy(const std::string& name)
-{
-	Mutex::ScopedLock lock(_mapMtx);
-
-	if (_pLoggerMap)
-	{
-		LoggerMap::iterator it = _pLoggerMap->find(name);
-		if (it != _pLoggerMap->end())
-		{
-			it->second->release();
-			_pLoggerMap->erase(it);
-		}
-	}
-}
-
-
 void Logger::names(std::vector<std::string>& names)
 {
-	Mutex::ScopedLock lock(_mapMtx);
+	std::lock_guard<std::mutex> lock(getLoggerMutex());

 	names.clear();
 	if (_pLoggerMap)
@ -393,6 +490,14 @@ void Logger::names(std::vector<std::string>& names)
 	}
 }

+Logger& Logger::unsafeCreate(const std::string & name, Channel * pChannel, int level)
+{
+	if (find(name)) throw ExistsException();
+	Logger* pLogger = new Logger(name, pChannel, level);
+	add(pLogger);
+
+	return *pLogger;
+}

 Logger& Logger::parent(const std::string& name)
 {
@ -478,7 +583,8 @@ void Logger::add(Logger* pLogger)
 {
 	if (!_pLoggerMap)
 		_pLoggerMap = new LoggerMap;
-	_pLoggerMap->insert(LoggerMap::value_type(pLogger->name(), pLogger));
+
+	_pLoggerMap->emplace(pLogger->name(), LoggerEntry{pLogger, false /*owned_by_shared_ptr*/});
 }


--- a/cmake/sanitize.cmake
+++ b/cmake/sanitize.cmake
@ -79,7 +79,10 @@ if (SANITIZE_COVERAGE)

    # But the actual coverage will be enabled on per-library basis: for ClickHouse code, but not for 3rd-party.
    set (COVERAGE_FLAGS "-fsanitize-coverage=trace-pc-guard,pc-table")
-endif()

-set (WITHOUT_COVERAGE_FLAGS "-fno-profile-instr-generate -fno-coverage-mapping -fno-sanitize-coverage=trace-pc-guard,pc-table")
-set (WITHOUT_COVERAGE_FLAGS_LIST -fno-profile-instr-generate -fno-coverage-mapping -fno-sanitize-coverage=trace-pc-guard,pc-table)
+    set (WITHOUT_COVERAGE_FLAGS "-fno-profile-instr-generate -fno-coverage-mapping -fno-sanitize-coverage=trace-pc-guard,pc-table")
+    set (WITHOUT_COVERAGE_FLAGS_LIST -fno-profile-instr-generate -fno-coverage-mapping -fno-sanitize-coverage=trace-pc-guard,pc-table)
+else()
+    set (WITHOUT_COVERAGE_FLAGS "")
+    set (WITHOUT_COVERAGE_FLAGS_LIST "")
+endif()
--- a/contrib/corrosion-cmake/CMakeLists.txt
+++ b/contrib/corrosion-cmake/CMakeLists.txt
@ -1,8 +1,5 @@
 if (NOT ENABLE_LIBRARIES)
  set(DEFAULT_ENABLE_RUST FALSE)
-elseif((CMAKE_TOOLCHAIN_FILE MATCHES "darwin") AND (CMAKE_TOOLCHAIN_FILE MATCHES "aarch64"))
-  message(STATUS "Rust is not available on aarch64-apple-darwin")
-  set(DEFAULT_ENABLE_RUST FALSE)
 else()
  list (APPEND CMAKE_MODULE_PATH "${ClickHouse_SOURCE_DIR}/contrib/corrosion/cmake")
  find_package(Rust)
@ -19,7 +16,9 @@ message(STATUS "Checking Rust toolchain for current target")

 # See https://doc.rust-lang.org/nightly/rustc/platform-support.html

-if((CMAKE_TOOLCHAIN_FILE MATCHES "linux/toolchain-x86_64") AND (CMAKE_TOOLCHAIN_FILE MATCHES "musl"))
+if(CMAKE_TOOLCHAIN_FILE MATCHES "ppc64le")
+    set(Rust_CARGO_TARGET "powerpc64le-unknown-linux-gnu")
+elseif((CMAKE_TOOLCHAIN_FILE MATCHES "linux/toolchain-x86_64") AND (CMAKE_TOOLCHAIN_FILE MATCHES "musl"))
    set(Rust_CARGO_TARGET "x86_64-unknown-linux-musl")
 elseif(CMAKE_TOOLCHAIN_FILE MATCHES "linux/toolchain-x86_64")
    set(Rust_CARGO_TARGET "x86_64-unknown-linux-gnu")
@ -29,14 +28,14 @@ elseif(CMAKE_TOOLCHAIN_FILE MATCHES "linux/toolchain-aarch64")
    set(Rust_CARGO_TARGET "aarch64-unknown-linux-gnu")
 elseif((CMAKE_TOOLCHAIN_FILE MATCHES "darwin") AND (CMAKE_TOOLCHAIN_FILE MATCHES "x86_64"))
    set(Rust_CARGO_TARGET "x86_64-apple-darwin")
+elseif((CMAKE_TOOLCHAIN_FILE MATCHES "darwin") AND (CMAKE_TOOLCHAIN_FILE MATCHES "aarch64"))
+    set(Rust_CARGO_TARGET "aarch64-apple-darwin")
 elseif((CMAKE_TOOLCHAIN_FILE MATCHES "freebsd") AND (CMAKE_TOOLCHAIN_FILE MATCHES "x86_64"))
    set(Rust_CARGO_TARGET "x86_64-unknown-freebsd")
 elseif(CMAKE_TOOLCHAIN_FILE MATCHES "linux/toolchain-riscv64")
    set(Rust_CARGO_TARGET "riscv64gc-unknown-linux-gnu")
-endif()
-
-if(CMAKE_TOOLCHAIN_FILE MATCHES "ppc64le")
-    set(Rust_CARGO_TARGET "powerpc64le-unknown-linux-gnu")
+else()
+    message(FATAL_ERROR "Unsupported rust target")
 endif()

 message(STATUS "Switched Rust target to ${Rust_CARGO_TARGET}")
--- a/contrib/libssh-cmake/CMakeLists.txt
+++ b/contrib/libssh-cmake/CMakeLists.txt
@ -1,4 +1,4 @@
-option (ENABLE_SSH "Enable support for SSH keys and protocol" ON)
+option (ENABLE_SSH "Enable support for SSH keys and protocol" ${ENABLE_LIBRARIES})

 if (NOT ENABLE_SSH)
    message(STATUS "Not using SSH")
--- a/contrib/simdjson
+++ b/contrib/simdjson
@ -1 +1 @@
-Subproject commit 1075e8609c4afa253162d441437af929c29e31bb
+Subproject commit 6060be2fdf62edf4a8f51a8b0883d57d09397b30
--- a/docker/test/fasttest/Dockerfile
+++ b/docker/test/fasttest/Dockerfile
@ -22,7 +22,7 @@ RUN apt-get update \
        zstd \
    --yes --no-install-recommends

-RUN pip3 install numpy scipy pandas Jinja2
+RUN pip3 install numpy==1.26.3 scipy==1.12.0 pandas==1.5.3 Jinja2==3.1.3

 ARG odbc_driver_url="https://github.com/ClickHouse/clickhouse-odbc/releases/download/v1.1.4.20200302/clickhouse-odbc-1.1.4-Linux.tar.gz"

--- a/docker/test/integration/runner/dockerd-entrypoint.sh
+++ b/docker/test/integration/runner/dockerd-entrypoint.sh
@ -23,13 +23,15 @@ if [ -f /sys/fs/cgroup/cgroup.controllers ]; then
        > /sys/fs/cgroup/cgroup.subtree_control
 fi

-# In case of test hung it is convenient to use pytest --pdb to debug it,
-# and on hung you can simply press Ctrl-C and it will spawn a python pdb,
-# but on SIGINT dockerd will exit, so ignore it to preserve the daemon.
-trap '' INT
 # Binding to an IP address without --tlsverify is deprecated. Startup is intentionally being slowed
 # unless --tls=false or --tlsverify=false is set
-dockerd --host=unix:///var/run/docker.sock --tls=false --host=tcp://0.0.0.0:2375 --default-address-pool base=172.17.0.0/12,size=24 &>/ClickHouse/tests/integration/dockerd.log &
+#
+# In case of test hung it is convenient to use pytest --pdb to debug it,
+# and on hung you can simply press Ctrl-C and it will spawn a python pdb,
+# but on SIGINT dockerd will exit, so we spawn new session to ignore SIGINT by
+# docker.
+# Note, that if you will run it via runner, it will send SIGINT to docker anyway.
+setsid dockerd --host=unix:///var/run/docker.sock --tls=false --host=tcp://0.0.0.0:2375 --default-address-pool base=172.17.0.0/12,size=24 &>/ClickHouse/tests/integration/dockerd.log &

 set +e
 reties=0
--- a/docker/test/stateless/run.sh
+++ b/docker/test/stateless/run.sh
@ -246,13 +246,19 @@ clickhouse-client -q "system flush logs" ||:
 stop_logs_replication

 # Try to get logs while server is running
-successfuly_saved=0
-for table in query_log zookeeper_log trace_log transactions_info_log
+failed_to_save_logs=0
+for table in query_log zookeeper_log trace_log transactions_info_log metric_log
 do
-    clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst || successfuly_saved=$((successfuly_saved+$?))
+    err=$( { clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst; } 2>&1 )
+    echo "$err"
+    [[ "0" != "${#err}"  ]] && failed_to_save_logs=1
    if [[ -n "$USE_DATABASE_REPLICATED" ]] && [[ "$USE_DATABASE_REPLICATED" -eq 1 ]]; then
-        clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.1.tsv.zst || successfuly_saved=$((successfuly_saved+$?))
-        clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.2.tsv.zst || successfuly_saved=$((successfuly_saved+$?))
+        err=$( { clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.1.tsv.zst; } 2>&1 )
+        echo "$err"
+        [[ "0" != "${#err}"  ]] && failed_to_save_logs=1
+        err=$( { clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.2.tsv.zst; } 2>&1 )
+        echo "$err"
+        [[ "0" != "${#err}"  ]] && failed_to_save_logs=1
    fi
 done

@ -277,7 +283,7 @@ fi


 # If server crashed dump system logs with clickhouse-local
-if [ $successfuly_saved -ne 0 ]; then
+if [ $failed_to_save_logs -ne 0 ]; then
    # Compress tables.
    #
    # NOTE:
@ -285,7 +291,7 @@ if [ $successfuly_saved -ne 0 ]; then
    #   directly
    # - even though ci auto-compress some files (but not *.tsv) it does this only
    #   for files >64MB, we want this files to be compressed explicitly
-    for table in query_log zookeeper_log trace_log transactions_info_log
+    for table in query_log zookeeper_log trace_log transactions_info_log metric_log
    do
        clickhouse-local "$data_path_config" --only-system-tables -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst ||:
        if [[ -n "$USE_DATABASE_REPLICATED" ]] && [[ "$USE_DATABASE_REPLICATED" -eq 1 ]]; then
--- a/docs/changelogs/v23.9.1.1854-stable.md
+++ b/docs/changelogs/v23.9.1.1854-stable.md
@ -11,6 +11,7 @@ sidebar_label: 2023
 * Remove the `status_info` configuration option and dictionaries status from the default Prometheus handler. [#54090](https://github.com/ClickHouse/ClickHouse/pull/54090) ([Alexey Milovidov](https://github.com/alexey-milovidov)).
 * The experimental parts metadata cache is removed from the codebase. [#54215](https://github.com/ClickHouse/ClickHouse/pull/54215) ([Alexey Milovidov](https://github.com/alexey-milovidov)).
 * Disable setting `input_format_json_try_infer_numbers_from_strings` by default, so we don't try to infer numbers from strings in JSON formats by default to avoid possible parsing errors when sample data contains strings that looks like a number. [#55099](https://github.com/ClickHouse/ClickHouse/pull/55099) ([Kruglov Pavel](https://github.com/Avogar)).
+* IPv6 bloom filter indexes created prior to March 2023 are not compatible with current version and have to be rebuilt. [#54200](https://github.com/ClickHouse/ClickHouse/pull/54200) ([Yakov Olkhovskiy](https://github.com/yakov-olkhovskiy)).

 #### New Feature
 * Added new type of authentication based on SSH keys. It works only for Native TCP protocol. [#41109](https://github.com/ClickHouse/ClickHouse/pull/41109) ([George Gamezardashvili](https://github.com/InfJoker)).
--- a/docs/en/engines/table-engines/mergetree-family/mergetree.md
+++ b/docs/en/engines/table-engines/mergetree-family/mergetree.md
@ -39,8 +39,8 @@ If you need to update rows frequently, we recommend using the [`ReplacingMergeTr
 ``` sql
 CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
 (
-    name1 [type1] [[NOT] NULL] [DEFAULT|MATERIALIZED|ALIAS|EPHEMERAL expr1] [COMMENT ...] [CODEC(codec1)] [STATISTIC(stat1)] [TTL expr1] [PRIMARY KEY],
-    name2 [type2] [[NOT] NULL] [DEFAULT|MATERIALIZED|ALIAS|EPHEMERAL expr2] [COMMENT ...] [CODEC(codec2)] [STATISTIC(stat2)] [TTL expr2] [PRIMARY KEY],
+    name1 [type1] [[NOT] NULL] [DEFAULT|MATERIALIZED|ALIAS|EPHEMERAL expr1] [COMMENT ...] [CODEC(codec1)] [STATISTIC(stat1)] [TTL expr1] [PRIMARY KEY] [SETTINGS (name = value, ...)],
+    name2 [type2] [[NOT] NULL] [DEFAULT|MATERIALIZED|ALIAS|EPHEMERAL expr2] [COMMENT ...] [CODEC(codec2)] [STATISTIC(stat2)] [TTL expr2] [PRIMARY KEY] [SETTINGS (name = value, ...)],
    ...
    INDEX index_name1 expr1 TYPE type1(...) [GRANULARITY value1],
    INDEX index_name2 expr2 TYPE type2(...) [GRANULARITY value2],
@ -56,7 +56,7 @@ ORDER BY expr
    [DELETE|TO DISK 'xxx'|TO VOLUME 'xxx' [, ...] ]
    [WHERE conditions]
    [GROUP BY key_expr [SET v1 = aggr_func(v1) [, v2 = aggr_func(v2) ...]] ] ]
-[SETTINGS name=value, ...]
+[SETTINGS name = value, ...]
 ```

 For a description of parameters, see the [CREATE query description](/docs/en/sql-reference/statements/create/table.md).
@ -620,7 +620,7 @@ The `TTL` clause can’t be used for key columns.
 #### Creating a table with `TTL`:

 ``` sql
-CREATE TABLE example_table
+CREATE TABLE tab
 (
    d DateTime,
    a Int TTL d + INTERVAL 1 MONTH,
@ -635,7 +635,7 @@ ORDER BY d;
 #### Adding TTL to a column of an existing table

 ``` sql
-ALTER TABLE example_table
+ALTER TABLE tab
    MODIFY COLUMN
    c String TTL d + INTERVAL 1 DAY;
 ```
@ -643,7 +643,7 @@ ALTER TABLE example_table
 #### Altering TTL of the column

 ``` sql
-ALTER TABLE example_table
+ALTER TABLE tab
    MODIFY COLUMN
    c String TTL d + INTERVAL 1 MONTH;
 ```
@ -681,7 +681,7 @@ If a column is not part of the `GROUP BY` expression and is not set explicitly i
 #### Creating a table with `TTL`:

 ``` sql
-CREATE TABLE example_table
+CREATE TABLE tab
 (
    d DateTime,
    a Int
@ -697,7 +697,7 @@ TTL d + INTERVAL 1 MONTH DELETE,
 #### Altering `TTL` of the table:

 ``` sql
-ALTER TABLE example_table
+ALTER TABLE tab
    MODIFY TTL d + INTERVAL 1 DAY;
 ```

@ -1366,7 +1366,7 @@ In this sample configuration:
 The statistic declaration is in the columns section of the `CREATE` query for tables from the `*MergeTree*` Family when we enable `set allow_experimental_statistic = 1`.

 ``` sql
-CREATE TABLE example_table
+CREATE TABLE tab
 (
    a Int64 STATISTIC(tdigest),
    b Float64
@ -1378,8 +1378,8 @@ ORDER BY a
 We can also manipulate statistics with `ALTER` statements.

 ```sql
-ALTER TABLE example_table ADD STATISTIC b TYPE tdigest;
-ALTER TABLE example_table DROP STATISTIC a TYPE tdigest;
+ALTER TABLE tab ADD STATISTIC b TYPE tdigest;
+ALTER TABLE tab DROP STATISTIC a TYPE tdigest;
 ```

 These lightweight statistics aggregate information about distribution of values in columns.
@ -1390,3 +1390,42 @@ They can be used for query optimization when we enable `set allow_statistic_opti
 -   `tdigest`

    Stores distribution of values from numeric columns in [TDigest](https://github.com/tdunning/t-digest) sketch.
+
+## Column-level Settings {#column-level-settings}
+
+Certain MergeTree settings can be override at column level:
+
+- `max_compress_block_size` — Maximum size of blocks of uncompressed data before compressing for writing to a table.
+- `min_compress_block_size` — Minimum size of blocks of uncompressed data required for compression when writing the next mark.
+
+Example:
+
+```sql
+CREATE TABLE tab
+(
+    id Int64,
+    document String SETTINGS (min_compress_block_size = 16777216, max_compress_block_size = 16777216)
+)
+ENGINE = MergeTree
+ORDER BY id
+```
+
+Column-level settings can be modified or removed using [ALTER MODIFY COLUMN](/docs/en/sql-reference/statements/alter/column.md), for example:
+
+- Remove `SETTINGS` from column declaration:
+
+```sql
+ALTER TABLE tab MODIFY COLUMN document REMOVE SETTINGS;
+```
+
+- Modify a setting:
+
+```sql
+ALTER TABLE tab MODIFY COLUMN document MODIFY SETTING min_compress_block_size = 8192;
+```
+
+- Reset one or more settings, also removes the setting declaration in the column expression of the table's CREATE query.
+
+```sql
+ALTER TABLE tab MODIFY COLUMN document RESET SETTING min_compress_block_size;
+```
--- a/docs/en/operations/server-configuration-parameters/settings.md
+++ b/docs/en/operations/server-configuration-parameters/settings.md
@ -2866,3 +2866,10 @@ This also allows a mix of resolver types can be used.
 ### disable_tunneling_for_https_requests_over_http_proxy {#disable_tunneling_for_https_requests_over_http_proxy}

 By default, tunneling (i.e, `HTTP CONNECT`) is used to make `HTTPS` requests over `HTTP` proxy. This setting can be used to disable it.
+
+## max_materialized_views_count_for_table {#max_materialized_views_count_for_table}
+
+A limit on the number of materialized views attached to a table.
+Note that only directly dependent views are considered here, and the creation of one view on top of another view is not considered.
+
+Default value: `0`.
--- a/docs/en/operations/settings/settings.md
+++ b/docs/en/operations/settings/settings.md
@ -2040,6 +2040,32 @@ SELECT * FROM test_table
 └───┘
 ```

+## update_insert_deduplication_token_in_dependent_materialized_views {#update-insert-deduplication-token-in-dependent-materialized-views}
+
+Allows to update `insert_deduplication_token` with table identifier during insert in dependent materialized views, if setting `deduplicate_blocks_in_dependent_materialized_views` is enabled and `insert_deduplication_token` is set.
+
+Possible values:
+
+      0 — Disabled.
+      1 — Enabled.
+
+Default value: 0.
+
+Usage:
+
+If setting `deduplicate_blocks_in_dependent_materialized_views` is enabled, `insert_deduplication_token` is passed to dependent materialized views. But in complex INSERT flows it is possible that we want to avoid deduplication for dependent materialized views.
+
+Example:
+```
+landing -┬--> mv_1_1 ---> ds_1_1 ---> mv_2_1 --┬-> ds_2_1 ---> mv_3_1 ---> ds_3_1
+         |                                     |
+         └--> mv_1_2 ---> ds_1_2 ---> mv_2_2 --┘
+```
+
+In this example we want to avoid deduplication for two different blocks generated from `mv_2_1` and `mv_2_2` that will be inserted into `ds_2_1`. Without `update_insert_deduplication_token_in_dependent_materialized_views` setting enabled, those two different blocks will be deduplicated, because different blocks from `mv_2_1` and `mv_2_2` will have the same `insert_deduplication_token`.
+
+If setting `update_insert_deduplication_token_in_dependent_materialized_views` is enabled, during each insert into dependent materialized views `insert_deduplication_token` is updated with table identifier, so block from `mv_2_1` and block from `mv_2_2` will have different `insert_deduplication_token` and will not be deduplicated.
+
 ## insert_keeper_max_retries

 The setting sets the maximum number of retries for ClickHouse Keeper (or ZooKeeper) requests during insert into replicated MergeTree. Only Keeper requests which failed due to network error, Keeper session timeout, or request timeout are considered for retries.
@ -5165,7 +5191,7 @@ SETTINGS(dictionary_use_async_executor=1, max_threads=8);
 ## storage_metadata_write_full_object_key {#storage_metadata_write_full_object_key}

 When set to `true` the metadata files are written with `VERSION_FULL_OBJECT_KEY` format version. With that format full object storage key names are written to the metadata files.
-When set to `false` the metadata files are written with the previous format version, `VERSION_INLINE_DATA`. With that format only suffixes of object storage key names are are written to the metadata files. The prefix for all of object storage key names is set in configurations files at `storage_configuration.disks` section. 
+When set to `false` the metadata files are written with the previous format version, `VERSION_INLINE_DATA`. With that format only suffixes of object storage key names are are written to the metadata files. The prefix for all of object storage key names is set in configurations files at `storage_configuration.disks` section.

 Default value: `false`.

@ -5176,6 +5202,95 @@ When set to `false` than all attempts are made with identical timeouts.

 Default value: `true`.

+## allow_experimental_variant_type {#allow_experimental_variant_type}
+
+Allows creation of experimental [Variant](../../sql-reference/data-types/variant.md).
+
+Default value: `false`.
+
+## use_variant_as_common_type {#use_variant_as_common_type}
+
+Allows to use `Variant` type as a result type for [if](../../sql-reference/functions/conditional-functions.md/#if)/[multiIf](../../sql-reference/functions/conditional-functions.md/#multiif)/[array](../../sql-reference/functions/array-functions.md)/[map](../../sql-reference/functions/tuple-map-functions.md) functions when there is no common type for argument types.
+
+Example:
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT toTypeName(if(number % 2, number, range(number))) as variant_type FROM numbers(1);
+SELECT if(number % 2, number, range(number)) as variant FROM numbers(5);
+```
+
+```text
+┌─variant_type───────────────────┐
+│ Variant(Array(UInt64), UInt64) │
+└────────────────────────────────┘
+┌─variant───┐
+│ []        │
+│ 1         │
+│ [0,1]     │
+│ 3         │
+│ [0,1,2,3] │
+└───────────┘
+```
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT toTypeName(multiIf((number % 4) = 0, 42, (number % 4) = 1, [1, 2, 3], (number % 4) = 2, 'Hello, World!', NULL)) AS variant_type FROM numbers(1);
+SELECT multiIf((number % 4) = 0, 42, (number % 4) = 1, [1, 2, 3], (number % 4) = 2, 'Hello, World!', NULL) AS variant FROM numbers(4);
+```
+
+```text
+─variant_type─────────────────────────┐
+│ Variant(Array(UInt8), String, UInt8) │
+└──────────────────────────────────────┘
+
+┌─variant───────┐
+│ 42            │
+│ [1,2,3]       │
+│ Hello, World! │
+│ ᴺᵁᴸᴸ          │
+└───────────────┘
+```
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT toTypeName(array(range(number), number, 'str_' || toString(number))) as array_of_variants_type from numbers(1);
+SELECT array(range(number), number, 'str_' || toString(number)) as array_of_variants FROM numbers(3);
+```
+
+```text
+┌─array_of_variants_type────────────────────────┐
+│ Array(Variant(Array(UInt64), String, UInt64)) │
+└───────────────────────────────────────────────┘
+
+┌─array_of_variants─┐
+│ [[],0,'str_0']    │
+│ [[0],1,'str_1']   │
+│ [[0,1],2,'str_2'] │
+└───────────────────┘
+```
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT toTypeName(map('a', range(number), 'b', number, 'c', 'str_' || toString(number))) as map_of_variants_type from numbers(1);
+SELECT map('a', range(number), 'b', number, 'c', 'str_' || toString(number)) as map_of_variants FROM numbers(3);
+```
+
+```text
+┌─map_of_variants_type────────────────────────────────┐
+│ Map(String, Variant(Array(UInt64), String, UInt64)) │
+└─────────────────────────────────────────────────────┘
+
+┌─map_of_variants───────────────┐
+│ {'a':[],'b':0,'c':'str_0'}    │
+│ {'a':[0],'b':1,'c':'str_1'}   │
+│ {'a':[0,1],'b':2,'c':'str_2'} │
+└───────────────────────────────┘
+```
+
+
+Default value: `false`.
+
 ## max_partition_size_to_drop

 Restriction on dropping partitions in query time.
@ -5197,3 +5312,13 @@ The value 0 means that you can delete all tables without any restrictions.
 :::note
 This query setting overwrites its server setting equivalent, see [max_table_size_to_drop](/docs/en/operations/server-configuration-parameters/settings.md/#max-table-size-to-drop)
 :::
+
+## iceberg_engine_ignore_schema_evolution {#iceberg_engine_ignore_schema_evolution}
+
+Allow to ignore schema evolution in Iceberg table engine and read all data using schema specified by the user on table creation or latest schema parsed from metadata on table creation.
+
+:::note
+Enabling this setting can lead to incorrect result as in case of evolved schema all data files will be read using the same schema.
+:::
+
+Default value: 'false'.
--- a/docs/en/operations/system-tables/asynchronous_metric_log.md
+++ b/docs/en/operations/system-tables/asynchronous_metric_log.md
@ -10,7 +10,7 @@ Columns:
 - `hostname` ([LowCardinality(String)](../../sql-reference/data-types/string.md)) — Hostname of the server executing the query.
 - `event_date` ([Date](../../sql-reference/data-types/date.md)) — Event date.
 - `event_time` ([DateTime](../../sql-reference/data-types/datetime.md)) — Event time.
- `name` ([String](../../sql-reference/data-types/string.md)) — Metric name.
+- `metric` ([String](../../sql-reference/data-types/string.md)) — Metric name.
 - `value` ([Float64](../../sql-reference/data-types/float.md)) — Metric value.

 **Example**
--- a/docs/en/operations/system-tables/metrics.md
+++ b/docs/en/operations/system-tables/metrics.md
@ -287,7 +287,7 @@ Number of threads in the HashedDictionary thread pool running a task.

 ### IOPrefetchThreads

-Number of threads in the IO prefertch thread pool.
+Number of threads in the IO prefetch thread pool.

 ### IOPrefetchThreadsActive

--- a/docs/en/sql-reference/data-types/float.md
+++ b/docs/en/sql-reference/data-types/float.md
@ -35,8 +35,8 @@ Types are equivalent to types of C:

 Aliases:

- `Float32` — `FLOAT`.
- `Float64` — `DOUBLE`.
+- `Float32` — `FLOAT`, `REAL`, `SINGLE`.
+- `Float64` — `DOUBLE`, `DOUBLE PRECISION`.

 When creating tables, numeric parameters for floating point numbers can be set (e.g. `FLOAT(12)`, `FLOAT(15, 22)`, `DOUBLE(12)`, `DOUBLE(4, 18)`), but ClickHouse ignores them.

--- a/docs/en/sql-reference/data-types/int-uint.md
+++ b/docs/en/sql-reference/data-types/int-uint.md
@ -21,10 +21,10 @@ When creating tables, numeric parameters for integer numbers can be set (e.g. `T

 Aliases:

- `Int8` — `TINYINT`, `BOOL`, `BOOLEAN`, `INT1`.
- `Int16` — `SMALLINT`, `INT2`.
- `Int32` — `INT`, `INT4`, `INTEGER`.
- `Int64` — `BIGINT`.
+- `Int8` — `TINYINT`, `INT1`, `BYTE`, `TINYINT SIGNED`, `INT1 SIGNED`.
+- `Int16` — `SMALLINT`, `SMALLINT SIGNED`.
+- `Int32` — `INT`, `INTEGER`, `MEDIUMINT`, `MEDIUMINT SIGNED`, `INT SIGNED`, `INTEGER SIGNED`.
+- `Int64` — `BIGINT`, `SIGNED`, `BIGINT SIGNED`, `TIME`.

 ## UInt Ranges

@ -34,3 +34,11 @@ Aliases:
 - `UInt64` — \[0 : 18446744073709551615\]
 - `UInt128` — \[0 : 340282366920938463463374607431768211455\]
 - `UInt256` — \[0 : 115792089237316195423570985008687907853269984665640564039457584007913129639935\]
+
+Aliases:
+
+- `UInt8` — `TINYINT UNSIGNED`, `INT1 UNSIGNED`.
+- `UInt16` — `SMALLINT UNSIGNED`.
+- `UInt32` — `MEDIUMINT UNSIGNED`, `INT UNSIGNED`, `INTEGER UNSIGNED`
+- `UInt64` — `UNSIGNED`, `BIGINT UNSIGNED`, `BIT`, `SET`
+
--- a/docs/en/sql-reference/data-types/variant.md
+++ b/docs/en/sql-reference/data-types/variant.md
@ -0,0 +1,245 @@
+---
+slug: /en/sql-reference/data-types/json
+sidebar_position: 55
+sidebar_label: Variant
+---
+
+# Variant(T1, T2, T3, ...)
+
+This type represents a union of other data types. Type `Variant(T1, T2, ..., TN)` means that each row of this type 
+has a value of either type `T1` or `T2` or ... or `TN` or none of them (`NULL` value).
+
+The order of nested types doesn't matter: Variant(T1, T2) = Variant(T2, T1).
+Nested types can be arbitrary types except Nullable(...), LowCardinality(Nullable(...)) and Variant(...) types.
+
+:::note
+The Variant data type is an experimental feature. To use it, set `allow_experimental_variant_type = 1`.
+:::
+
+## Creating Variant
+
+Using `Variant` type in table column definition:
+
+```sql
+CREATE TABLE test (v Variant(UInt64, String, Array(UInt64))) ENGINE = Memory;
+INSERT INTO test VALUES (NULL), (42), ('Hello, World!'), ([1, 2, 3]);
+SELECT v FROM test;
+```
+
+```text
+┌─v─────────────┐
+│ ᴺᵁᴸᴸ          │
+│ 42            │
+│ Hello, World! │
+│ [1,2,3]       │
+└───────────────┘
+```
+
+Using CAST from ordinary columns:
+
+```sql
+SELECT toTypeName(variant) as type_name, 'Hello, World!'::Variant(UInt64, String, Array(UInt64)) as variant;
+```
+
+```text
+┌─type_name──────────────────────────────┬─variant───────┐
+│ Variant(Array(UInt64), String, UInt64) │ Hello, World! │
+└────────────────────────────────────────┴───────────────┘
+```
+
+Using functions `if/multiIf` when arguments don't have common type (setting `use_variant_as_common_type` should be enabled for it):
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT if(number % 2, number, range(number)) as variant FROM numbers(5);
+```
+
+```text
+┌─variant───┐
+│ []        │
+│ 1         │
+│ [0,1]     │
+│ 3         │
+│ [0,1,2,3] │
+└───────────┘
+```
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT multiIf((number % 4) = 0, 42, (number % 4) = 1, [1, 2, 3], (number % 4) = 2, 'Hello, World!', NULL) AS variant FROM numbers(4);
+```
+
+```text
+┌─variant───────┐
+│ 42            │
+│ [1,2,3]       │
+│ Hello, World! │
+│ ᴺᵁᴸᴸ          │
+└───────────────┘
+```
+
+Using functions 'array/map' if array elements/map values don't have common type (setting `use_variant_as_common_type` should be enabled for it):
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT array(range(number), number, 'str_' || toString(number)) as array_of_variants FROM numbers(3);
+```
+
+```text
+┌─array_of_variants─┐
+│ [[],0,'str_0']    │
+│ [[0],1,'str_1']   │
+│ [[0,1],2,'str_2'] │
+└───────────────────┘
+```
+
+```sql
+SET use_variant_as_common_type = 1;
+SELECT map('a', range(number), 'b', number, 'c', 'str_' || toString(number)) as map_of_variants FROM numbers(3);
+```
+
+```text
+┌─map_of_variants───────────────┐
+│ {'a':[],'b':0,'c':'str_0'}    │
+│ {'a':[0],'b':1,'c':'str_1'}   │
+│ {'a':[0,1],'b':2,'c':'str_2'} │
+└───────────────────────────────┘
+```
+
+## Reading Variant nested types as subcolumns
+
+Variant type supports reading a single nested type from a Variant column using the type name as a subcolumn.
+So, if you have column `variant Variant(T1, T2, T3)` you can read a subcolumn of type `T2` using syntax `variant.T2`,
+this subcolumn will have type `Nullable(T2)` if `T2` can be inside `Nullable` and `T2` otherwise. This subcolumn will
+be the same size as original `Variant` column and will contain `NULL` values (or empty values if `T2` cannot be inside `Nullable`)
+in all rows in which original `Variant` column doesn't have type `T2`.
+
+Variant subcolumns can be also read using function `variantElement(variant_column, type_name)`. 
+
+Examples:
+
+```sql
+CREATE TABLE test (v Variant(UInt64, String, Array(UInt64))) ENGINE = Memory;
+INSERT INTO test VALUES (NULL), (42), ('Hello, World!'), ([1, 2, 3]);
+SELECT v, v.String, v.UInt64, v.`Array(UInt64)` FROM test;
+```
+
+```text
+┌─v─────────────┬─v.String──────┬─v.UInt64─┬─v.Array(UInt64)─┐
+│ ᴺᵁᴸᴸ          │ ᴺᵁᴸᴸ          │     ᴺᵁᴸᴸ │ []              │
+│ 42            │ ᴺᵁᴸᴸ          │       42 │ []              │
+│ Hello, World! │ Hello, World! │     ᴺᵁᴸᴸ │ []              │
+│ [1,2,3]       │ ᴺᵁᴸᴸ          │     ᴺᵁᴸᴸ │ [1,2,3]         │
+└───────────────┴───────────────┴──────────┴─────────────────┘
+```
+
+```sql
+SELECT toTypeName(v.String), toTypeName(v.UInt64), toTypeName(v.`Array(UInt64)`) FROM test LIMIT 1;
+```
+
+```text
+┌─toTypeName(v.String)─┬─toTypeName(v.UInt64)─┬─toTypeName(v.Array(UInt64))─┐
+│ Nullable(String)     │ Nullable(UInt64)     │ Array(UInt64)               │
+└──────────────────────┴──────────────────────┴─────────────────────────────┘
+```
+
+```sql
+SELECT v, variantElement(v, 'String'), variantElement(v, 'UInt64'), variantElement(v, 'Array(UInt64)') FROM test;
+```
+
+```text
+┌─v─────────────┬─variantElement(v, 'String')─┬─variantElement(v, 'UInt64')─┬─variantElement(v, 'Array(UInt64)')─┐
+│ ᴺᵁᴸᴸ          │ ᴺᵁᴸᴸ                        │                        ᴺᵁᴸᴸ │ []                                 │
+│ 42            │ ᴺᵁᴸᴸ                        │                          42 │ []                                 │
+│ Hello, World! │ Hello, World!               │                        ᴺᵁᴸᴸ │ []                                 │
+│ [1,2,3]       │ ᴺᵁᴸᴸ                        │                        ᴺᵁᴸᴸ │ [1,2,3]                            │
+└───────────────┴─────────────────────────────┴─────────────────────────────┴────────────────────────────────────┘
+```
+
+## Conversion between Variant column and other columns
+
+There are 3 possible conversions that can be performed with Variant column.
+
+### Converting an ordinary column to a Variant column
+
+It is possible to convert ordinary column with type `T` to a `Variant` column containing this type:
+
+```sql
+SELECT toTypeName(variant) as type_name, 'Hello, World!'::Variant(UInt64, String, Array(UInt64)) as variant;
+```
+
+```text
+┌─type_name──────────────────────────────┬─variant───────┐
+│ Variant(Array(UInt64), String, UInt64) │ Hello, World! │
+└────────────────────────────────────────┴───────────────┘
+```
+
+### Converting a Variant column to an ordinary column
+
+It is possible to convert a `Variant` column to an ordinary column. In this case all nested variants will be converted to a destination type:
+
+```sql
+CREATE TABLE test (v Variant(UInt64, String)) ENGINE = Memory;
+INSERT INTO test VALUES (NULL), (42), ('42.42');
+SELECT v::Nullable(Float64) FROM test;
+```
+
+```text
+┌─CAST(v, 'Nullable(Float64)')─┐
+│                         ᴺᵁᴸᴸ │
+│                           42 │
+│                        42.42 │
+└──────────────────────────────┘
+```
+
+### Converting a Variant to another Variant
+
+It is possible to convert a `Variant` column to another `Variant` column, but only if the destination `Variant` column contains all nested types from the original `Variant`:
+
+```sql
+CREATE TABLE test (v Variant(UInt64, String)) ENGINE = Memory;
+INSERT INTO test VALUES (NULL), (42), ('String');
+SELECT v::Variant(UInt64, String, Array(UInt64)) FROM test;
+```
+
+```text
+┌─CAST(v, 'Variant(UInt64, String, Array(UInt64))')─┐
+│ ᴺᵁᴸᴸ                                              │
+│ 42                                                │
+│ String                                            │
+└───────────────────────────────────────────────────┘
+```
+
+
+## Reading Variant type from the data
+
+All text formats (TSV, CSV, CustomSeparated, Values, JSONEachRow, etc) supports reading `Variant` type. During data parsing ClickHouse tries to insert value into most appropriate variant type.
+
+Example:
+
+```sql
+SELECT
+    v,
+    variantElement(v, 'String') AS str,
+    variantElement(v, 'UInt64') AS num,
+    variantElement(v, 'Float64') AS float,
+    variantElement(v, 'DateTime') AS date,
+    variantElement(v, 'Array(UInt64)') AS arr
+FROM format(JSONEachRow, 'v Variant(String, UInt64, Float64, DateTime, Array(UInt64))', $$
+{"v" : "Hello, World!"},
+{"v" : 42},
+{"v" : 42.42},
+{"v" : "2020-01-01 00:00:00"},
+{"v" : [1, 2, 3]}
+$$)
+```
+
+```text
+┌─v───────────────────┬─str───────────┬──num─┬─float─┬────────────────date─┬─arr─────┐
+│ Hello, World!       │ Hello, World! │ ᴺᵁᴸᴸ │  ᴺᵁᴸᴸ │                ᴺᵁᴸᴸ │ []      │
+│ 42                  │ ᴺᵁᴸᴸ          │   42 │  ᴺᵁᴸᴸ │                ᴺᵁᴸᴸ │ []      │
+│ 42.42               │ ᴺᵁᴸᴸ          │ ᴺᵁᴸᴸ │ 42.42 │                ᴺᵁᴸᴸ │ []      │
+│ 2020-01-01 00:00:00 │ ᴺᵁᴸᴸ          │ ᴺᵁᴸᴸ │  ᴺᵁᴸᴸ │ 2020-01-01 00:00:00 │ []      │
+│ [1,2,3]             │ ᴺᵁᴸᴸ          │ ᴺᵁᴸᴸ │  ᴺᵁᴸᴸ │                ᴺᵁᴸᴸ │ [1,2,3] │
+└─────────────────────┴───────────────┴──────┴───────┴─────────────────────┴─────────┘
+```
--- a/docs/en/sql-reference/dictionaries/index.md
+++ b/docs/en/sql-reference/dictionaries/index.md
@ -1805,6 +1805,7 @@ Example of settings:
 ``` xml
 <source>
  <postgresql>
+      <host>postgresql-hostname</hoat>
      <port>5432</port>
      <user>clickhouse</user>
      <password>qwerty</password>
--- a/docs/en/sql-reference/functions/other-functions.md
+++ b/docs/en/sql-reference/functions/other-functions.md
@ -2832,6 +2832,43 @@ Result:
 └─────────────────────────────────────────────────────────────────────────┘
 ```

+## variantElement
+
+Extracts a column with specified type from a `Variant` column.
+
+**Syntax**
+
+```sql
+variantElement(variant, type_name, [, default_value])
+```
+
+**Arguments**
+
+- `variant` — Variant column. [Variant](../../sql-reference/data-types/variant.md).
+- `type_name` — The name of the variant type to extract. [String](../../sql-reference/data-types/string.md).
+- `default_value` - The default value that will be used if variant doesn't have variant with specified type. Can be any type. Optional.
+
+**Returned value**
+
+- Subcolumn of a `Variant` column with specified type.
+
+**Example**
+
+```sql
+CREATE TABLE test (v Variant(UInt64, String, Array(UInt64))) ENGINE = Memory;
+INSERT INTO test VALUES (NULL), (42), ('Hello, World!'), ([1, 2, 3]);
+SELECT v, variantElement(v, 'String'), variantElement(v, 'UInt64'), variantElement(v, 'Array(UInt64)') FROM test;
+```
+
+```text
+┌─v─────────────┬─variantElement(v, 'String')─┬─variantElement(v, 'UInt64')─┬─variantElement(v, 'Array(UInt64)')─┐
+│ ᴺᵁᴸᴸ          │ ᴺᵁᴸᴸ                        │                        ᴺᵁᴸᴸ │ []                                 │
+│ 42            │ ᴺᵁᴸᴸ                        │                          42 │ []                                 │
+│ Hello, World! │ Hello, World!               │                        ᴺᵁᴸᴸ │ []                                 │
+│ [1,2,3]       │ ᴺᵁᴸᴸ                        │                        ᴺᵁᴸᴸ │ [1,2,3]                            │
+└───────────────┴─────────────────────────────┴─────────────────────────────┴────────────────────────────────────┘
+```
+
 ## minSampleSizeConversion

 Calculates minimum required sample size for an A/B test comparing conversions (proportions) in two samples.
--- a/docs/en/sql-reference/functions/time-series-functions.md
+++ b/docs/en/sql-reference/functions/time-series-functions.md
@ -77,8 +77,8 @@ The number of data points in `series` should be at least twice the value of `per

 **Returned value**

- An array of three arrays where the first array include seasonal components, the second array - trend,
-and the third array - residue component.
+- An array of four arrays where the first array include seasonal components, the second array - trend,
+the third array - residue component, and the fourth array - baseline(seasonal + trend) component.

 Type: [Array](../../sql-reference/data-types/array.md).

@ -107,6 +107,10 @@ Result:
    [
        0, 0.0000019073486, -0.0000019073486, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, -0.0000019073486, 0,
        0
+    ],
+    [
+        10.1, 20.449999, 40.340004, 10.100001, 20.45, 40.34, 10.100001, 20.45, 40.34, 10.1, 20.45, 40.34,
+        10.1, 20.45, 40.34, 10.1, 20.45, 40.34, 10.1, 20.45, 40.34, 10.100002, 20.45, 40.34
    ]]                                                                                                                   │
 └────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
 ```
--- a/docs/en/sql-reference/statements/alter/column.md
+++ b/docs/en/sql-reference/statements/alter/column.md
@ -23,10 +23,11 @@ The following actions are supported:
 - [RENAME COLUMN](#rename-column) — Renames an existing column.
 - [CLEAR COLUMN](#clear-column) — Resets column values.
 - [COMMENT COLUMN](#comment-column) — Adds a text comment to the column.
- [MODIFY COLUMN](#modify-column) — Changes column’s type, default expression and TTL.
+- [MODIFY COLUMN](#modify-column) — Changes column’s type, default expression, TTL, and column settings.
 - [MODIFY COLUMN REMOVE](#modify-column-remove) — Removes one of the column properties.
+- [MODIFY COLUMN MODIFY SETTING](#modify-column-modify-setting) - Changes column settings.
+- [MODIFY COLUMN RESET SETTING](#modify-column-reset-setting) - Reset column settings.
 - [MATERIALIZE COLUMN](#materialize-column) — Materializes the column in the parts where the column is missing.
-
 These actions are described in detail below.

 ## ADD COLUMN
@ -75,7 +76,7 @@ Deletes the column with the name `name`. If the `IF EXISTS` clause is specified,

 Deletes data from the file system. Since this deletes entire files, the query is completed almost instantly.

-:::tip    
+:::tip
 You can’t delete a column if it is referenced by [materialized view](/docs/en/sql-reference/statements/create/view.md/#materialized). Otherwise, it returns an error.
 :::

@ -208,7 +209,7 @@ The `ALTER` query for changing columns is replicated. The instructions are saved

 ## MODIFY COLUMN REMOVE

-Removes one of the column properties: `DEFAULT`, `ALIAS`, `MATERIALIZED`, `CODEC`, `COMMENT`, `TTL`.
+Removes one of the column properties: `DEFAULT`, `ALIAS`, `MATERIALIZED`, `CODEC`, `COMMENT`, `TTL`, `SETTING`.

 Syntax:

@ -228,6 +229,43 @@ ALTER TABLE table_with_ttl MODIFY COLUMN column_ttl REMOVE TTL;

 - [REMOVE TTL](ttl.md).

+
+## MODIFY COLUMN MODIFY SETTING
+
+Modify a column setting.
+
+Syntax:
+
+```sql
+ALTER TABLE table_name MODIFY COLUMN MODIFY SETTING name=value,...;
+```
+
+**Example**
+
+Modify column's `max_compress_block_size` to `1MB`:
+
+```sql
+ALTER TABLE table_name MODIFY COLUMN MODIFY SETTING max_compress_block_size = 1048576;
+```
+
+## MODIFY COLUMN RESET SETTING
+
+Reset a column setting, also removes the setting declaration in the column expression of the table's CREATE query.
+
+Syntax:
+
+```sql
+ALTER TABLE table_name MODIFY COLUMN RESET SETTING name,...;
+```
+
+**Example**
+
+Remove column setting `max_compress_block_size` to `1MB`:
+
+```sql
+ALTER TABLE table_name MODIFY COLUMN REMOVE SETTING max_compress_block_size;
+```
+
 ## MATERIALIZE COLUMN

 Materializes or updates a column with an expression for a default value (`DEFAULT` or `MATERIALIZED`).
--- a/docs/en/sql-reference/statements/alter/view.md
+++ b/docs/en/sql-reference/statements/alter/view.md
@ -8,8 +8,6 @@ sidebar_label: VIEW

 You can modify `SELECT` query that was specified when a [materialized view](../create/view.md#materialized) was created with the `ALTER TABLE … MODIFY QUERY` statement without interrupting ingestion process.

-The `allow_experimental_alter_materialized_view_structure` setting must be enabled.
-
 This command is created to change materialized view created with `TO [db.]name` clause. It does not change the structure of the underling storage table and it does not change the columns' definition of the materialized view, because of this the application of this command is very limited for materialized views are created without `TO [db.]name` clause.

 **Example with TO table**
--- a/docs/en/sql-reference/statements/create/view.md
+++ b/docs/en/sql-reference/statements/create/view.md
@ -97,7 +97,7 @@ This feature is deprecated and will be removed in the future.

 For your convenience, the old documentation is located [here](https://pastila.nl/?00f32652/fdf07272a7b54bda7e13b919264e449f.md)

-## Refreshable Materialized View {#refreshable-materialized-view}
+## Refreshable Materialized View [Experimental] {#refreshable-materialized-view}

 ```sql
 CREATE MATERIALIZED VIEW [IF NOT EXISTS] [db.]table_name
@ -120,7 +120,8 @@ Differences from regular non-refreshable materialized views:

 :::note
 Refreshable materialized views are a work in progress. Setting `allow_experimental_refreshable_materialized_view = 1` is required for creating one. Current limitations:
- * not compatible with Replicated database or table engines,
+ * not compatible with Replicated database or table engines
+ * It is not supported in ClickHouse Cloud
 * require [Atomic database engine](../../../engines/database-engines/atomic.md),
 * no retries for failed refresh - we just skip to the next scheduled refresh time,
 * no limit on number of concurrent refreshes.
--- a/docs/en/sql-reference/statements/rename.md
+++ b/docs/en/sql-reference/statements/rename.md
@ -9,10 +9,6 @@ sidebar_label: RENAME
 Renames databases, tables, or dictionaries. Several entities can be renamed in a single query.
 Note that the `RENAME` query with several entities is non-atomic operation. To swap entities names atomically, use the [EXCHANGE](./exchange.md) statement.

-:::note    
-The `RENAME` query is supported by the [Atomic](../../engines/database-engines/atomic.md) database engine only.
-:::
-
 **Syntax**

 ```sql
--- a/docs/en/sql-reference/table-functions/executable.md
+++ b/docs/en/sql-reference/table-functions/executable.md
@ -7,7 +7,7 @@ keywords: [udf, user defined function, clickhouse, executable, table, function]

 # executable Table Function for UDFs

-The `executable` table function creates a table based on the output of a user-defined function (UDF) that you define in a script that outputs rows to **stdout**. The executable script is stored in the `users_scripts` directory and can read data from any source.
+The `executable` table function creates a table based on the output of a user-defined function (UDF) that you define in a script that outputs rows to **stdout**. The executable script is stored in the `users_scripts` directory and can read data from any source. Make sure your ClickHouse server has all the required packages to run the executable script. For example, if it is a Python script, ensure that the server has the necessary Python packages installed.

 You can optionally include one or more input queries that stream their results to **stdin** for the script to read.

--- a/programs/copier/ClusterCopier.h
+++ b/programs/copier/ClusterCopier.h
@ -20,7 +20,7 @@ public:
                  const String & host_id_,
                  const String & proxy_database_name_,
                  ContextMutablePtr context_,
-                  Poco::Logger * log_)
+                  LoggerRawPtr log_)
            : WithMutableContext(context_),
            task_zookeeper_path(task_path_),
            host_id(host_id_),
@ -230,7 +230,7 @@ private:

    bool experimental_use_sample_offset{false};

-    Poco::Logger * log;
+    LoggerRawPtr log;

    UInt64 max_table_tries = 3;
    UInt64 max_shard_partition_tries = 3;
--- a/programs/copier/ZooKeeperStaff.h
+++ b/programs/copier/ZooKeeperStaff.h
@ -177,7 +177,7 @@ public:
        auto watch_callback =
                [my_stale = stale] (const Coordination::WatchResponse & rsp)
                {
-                    auto logger = &Poco::Logger::get("ClusterCopier");
+                    auto logger = getLogger("ClusterCopier");
                    if (rsp.error == Coordination::Error::ZOK)
                    {
                        switch (rsp.type)
--- a/programs/disks/DisksApp.cpp
+++ b/programs/disks/DisksApp.cpp
@ -160,7 +160,7 @@ int DisksApp::main(const std::vector<String> & /*args*/)
    }
    else
    {
-        throw Exception(ErrorCodes::BAD_ARGUMENTS, "No config-file specifiged");
+        throw Exception(ErrorCodes::BAD_ARGUMENTS, "No config-file specified");
    }

    if (config().has("save-logs"))
--- a/programs/keeper-client/KeeperClient.cpp
+++ b/programs/keeper-client/KeeperClient.cpp
@ -375,7 +375,7 @@ int KeeperClient::main(const std::vector<String> & /* args */)

    if (!config().has("host") && !config().has("port") && !keys.empty())
    {
-        LOG_INFO(&Poco::Logger::get("KeeperClient"), "Found keeper node in the config.xml, will use it for connection");
+        LOG_INFO(getLogger("KeeperClient"), "Found keeper node in the config.xml, will use it for connection");

        for (const auto & key : keys)
        {
--- a/programs/keeper-converter/KeeperConverter.cpp
+++ b/programs/keeper-converter/KeeperConverter.cpp
@ -28,7 +28,7 @@ int mainEntryClickHouseKeeperConverter(int argc, char ** argv)
    po::store(po::command_line_parser(argc, argv).options(desc).run(), options);
    Poco::AutoPtr<Poco::ConsoleChannel> console_channel(new Poco::ConsoleChannel);

-    Poco::Logger * logger = &Poco::Logger::get("KeeperConverter");
+    LoggerPtr logger = getLogger("KeeperConverter");
    logger->setChannel(console_channel);

    if (options.count("help"))
--- a/programs/keeper/Keeper.cpp
+++ b/programs/keeper/Keeper.cpp
@ -624,7 +624,7 @@ catch (...)

 void Keeper::logRevision() const
 {
-    LOG_INFO(&Poco::Logger::get("Application"),
+    LOG_INFO(getLogger("Application"),
        "Starting ClickHouse Keeper {} (revision: {}, git hash: {}, build id: {}), PID {}",
        VERSION_STRING,
        ClickHouseRevision::getVersionRevision(),
--- a/programs/library-bridge/CatBoostLibraryHandlerFactory.cpp
+++ b/programs/library-bridge/CatBoostLibraryHandlerFactory.cpp
@ -13,7 +13,7 @@ CatBoostLibraryHandlerFactory & CatBoostLibraryHandlerFactory::instance()
 }

 CatBoostLibraryHandlerFactory::CatBoostLibraryHandlerFactory()
-    : log(&Poco::Logger::get("CatBoostLibraryHandlerFactory"))
+    : log(getLogger("CatBoostLibraryHandlerFactory"))
 {
 }

--- a/programs/library-bridge/CatBoostLibraryHandlerFactory.h
+++ b/programs/library-bridge/CatBoostLibraryHandlerFactory.h
@ -31,7 +31,7 @@ private:
    /// map: model path --> catboost library handler
    std::unordered_map<String, CatBoostLibraryHandlerPtr> library_handlers TSA_GUARDED_BY(mutex);
    std::mutex mutex;
-    Poco::Logger * log;
+    LoggerPtr log;
 };

 }
--- a/programs/library-bridge/ExternalDictionaryLibraryAPI.cpp
+++ b/programs/library-bridge/ExternalDictionaryLibraryAPI.cpp
@ -9,40 +9,40 @@ const char DICT_LOGGER_NAME[] = "LibraryDictionarySourceExternal";

 void ExternalDictionaryLibraryAPI::log(LogLevel level, CString msg)
 {
-    auto & logger = Poco::Logger::get(DICT_LOGGER_NAME);
+    auto logger = getLogger(DICT_LOGGER_NAME);
    switch (level)
    {
        case LogLevel::TRACE:
-            if (logger.trace())
-                logger.trace(msg);
+            if (logger->trace())
+                logger->trace(msg);
            break;
        case LogLevel::DEBUG:
-            if (logger.debug())
-                logger.debug(msg);
+            if (logger->debug())
+                logger->debug(msg);
            break;
        case LogLevel::INFORMATION:
-            if (logger.information())
-                logger.information(msg);
+            if (logger->information())
+                logger->information(msg);
            break;
        case LogLevel::NOTICE:
-            if (logger.notice())
-                logger.notice(msg);
+            if (logger->notice())
+                logger->notice(msg);
            break;
        case LogLevel::WARNING:
-            if (logger.warning())
-                logger.warning(msg);
+            if (logger->warning())
+                logger->warning(msg);
            break;
        case LogLevel::ERROR:
-            if (logger.error())
-                logger.error(msg);
+            if (logger->error())
+                logger->error(msg);
            break;
        case LogLevel::CRITICAL:
-            if (logger.critical())
-                logger.critical(msg);
+            if (logger->critical())
+                logger->critical(msg);
            break;
        case LogLevel::FATAL:
-            if (logger.fatal())
-                logger.fatal(msg);
+            if (logger->fatal())
+                logger->fatal(msg);
            break;
    }
 }
--- a/programs/library-bridge/ExternalDictionaryLibraryHandlerFactory.cpp
+++ b/programs/library-bridge/ExternalDictionaryLibraryHandlerFactory.cpp
@ -26,7 +26,7 @@ void ExternalDictionaryLibraryHandlerFactory::create(

    if (library_handlers.contains(dictionary_id))
    {
-        LOG_WARNING(&Poco::Logger::get("ExternalDictionaryLibraryHandlerFactory"), "Library handler with dictionary id {} already exists", dictionary_id);
+        LOG_WARNING(getLogger("ExternalDictionaryLibraryHandlerFactory"), "Library handler with dictionary id {} already exists", dictionary_id);
        return;
    }

--- a/programs/library-bridge/LibraryBridgeHandlerFactory.cpp
+++ b/programs/library-bridge/LibraryBridgeHandlerFactory.cpp
@ -12,7 +12,7 @@ LibraryBridgeHandlerFactory::LibraryBridgeHandlerFactory(
    size_t keep_alive_timeout_,
    ContextPtr context_)
    : WithContext(context_)
-    , log(&Poco::Logger::get(name_))
+    , log(getLogger(name_))
    , name(name_)
    , keep_alive_timeout(keep_alive_timeout_)
 {
--- a/programs/library-bridge/LibraryBridgeHandlerFactory.h
+++ b/programs/library-bridge/LibraryBridgeHandlerFactory.h
@ -19,7 +19,7 @@ public:
    std::unique_ptr<HTTPRequestHandler> createRequestHandler(const HTTPServerRequest & request) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;
    const std::string name;
    const size_t keep_alive_timeout;
 };
--- a/programs/library-bridge/LibraryBridgeHandlers.cpp
+++ b/programs/library-bridge/LibraryBridgeHandlers.cpp
@ -47,7 +47,7 @@ namespace
        if (!response.sent())
            *response.send() << message << '\n';

-        LOG_WARNING(&Poco::Logger::get("LibraryBridge"), fmt::runtime(message));
+        LOG_WARNING(getLogger("LibraryBridge"), fmt::runtime(message));
    }

    std::shared_ptr<Block> parseColumns(String && column_string)
@ -92,7 +92,7 @@ static void writeData(Block data, OutputFormatPtr format)
 ExternalDictionaryLibraryBridgeRequestHandler::ExternalDictionaryLibraryBridgeRequestHandler(size_t keep_alive_timeout_, ContextPtr context_)
    : WithContext(context_)
    , keep_alive_timeout(keep_alive_timeout_)
-    , log(&Poco::Logger::get("ExternalDictionaryLibraryBridgeRequestHandler"))
+    , log(getLogger("ExternalDictionaryLibraryBridgeRequestHandler"))
 {
 }

@ -380,7 +380,7 @@ void ExternalDictionaryLibraryBridgeRequestHandler::handleRequest(HTTPServerRequ
 ExternalDictionaryLibraryBridgeExistsHandler::ExternalDictionaryLibraryBridgeExistsHandler(size_t keep_alive_timeout_, ContextPtr context_)
    : WithContext(context_)
    , keep_alive_timeout(keep_alive_timeout_)
-    , log(&Poco::Logger::get("ExternalDictionaryLibraryBridgeExistsHandler"))
+    , log(getLogger("ExternalDictionaryLibraryBridgeExistsHandler"))
 {
 }

@ -419,7 +419,7 @@ CatBoostLibraryBridgeRequestHandler::CatBoostLibraryBridgeRequestHandler(
    size_t keep_alive_timeout_, ContextPtr context_)
    : WithContext(context_)
    , keep_alive_timeout(keep_alive_timeout_)
-    , log(&Poco::Logger::get("CatBoostLibraryBridgeRequestHandler"))
+    , log(getLogger("CatBoostLibraryBridgeRequestHandler"))
 {
 }

@ -623,7 +623,7 @@ void CatBoostLibraryBridgeRequestHandler::handleRequest(HTTPServerRequest & requ
 CatBoostLibraryBridgeExistsHandler::CatBoostLibraryBridgeExistsHandler(size_t keep_alive_timeout_, ContextPtr context_)
    : WithContext(context_)
    , keep_alive_timeout(keep_alive_timeout_)
-    , log(&Poco::Logger::get("CatBoostLibraryBridgeExistsHandler"))
+    , log(getLogger("CatBoostLibraryBridgeExistsHandler"))
 {
 }

--- a/programs/library-bridge/LibraryBridgeHandlers.h
+++ b/programs/library-bridge/LibraryBridgeHandlers.h
@ -26,7 +26,7 @@ private:
    static constexpr inline auto FORMAT = "RowBinary";

    const size_t keep_alive_timeout;
-    Poco::Logger * log;
+    LoggerPtr log;
 };


@ -40,7 +40,7 @@ public:

 private:
    const size_t keep_alive_timeout;
-    Poco::Logger * log;
+    LoggerPtr log;
 };


@ -69,7 +69,7 @@ public:

 private:
    const size_t keep_alive_timeout;
-    Poco::Logger * log;
+    LoggerPtr log;
 };


@ -83,7 +83,7 @@ public:

 private:
    const size_t keep_alive_timeout;
-    Poco::Logger * log;
+    LoggerPtr log;
 };

 }
--- a/programs/local/LocalServer.cpp
+++ b/programs/local/LocalServer.cpp
@ -221,7 +221,7 @@ void LocalServer::tryInitPath()
    {
        // The path is not provided explicitly - use a unique path in the system temporary directory
        // (or in the current dir if temporary don't exist)
-        Poco::Logger * log = &logger();
+        LoggerRawPtr log = &logger();
        std::filesystem::path parent_folder;
        std::filesystem::path default_path;

@ -631,7 +631,7 @@ void LocalServer::processConfig()

    tryInitPath();

-    Poco::Logger * log = &logger();
+    LoggerRawPtr log = &logger();

    /// Maybe useless
    if (config().has("macros"))
--- a/programs/odbc-bridge/ColumnInfoHandler.h
+++ b/programs/odbc-bridge/ColumnInfoHandler.h
@ -18,7 +18,7 @@ class ODBCColumnsInfoHandler : public HTTPRequestHandler, WithContext
 public:
    ODBCColumnsInfoHandler(size_t keep_alive_timeout_, ContextPtr context_)
        : WithContext(context_)
-        , log(&Poco::Logger::get("ODBCColumnsInfoHandler"))
+        , log(getLogger("ODBCColumnsInfoHandler"))
        , keep_alive_timeout(keep_alive_timeout_)
    {
    }
@ -26,7 +26,7 @@ public:
    void handleRequest(HTTPServerRequest & request, HTTPServerResponse & response, const ProfileEvents::Event & write_event) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;
    size_t keep_alive_timeout;
 };

--- a/programs/odbc-bridge/IdentifierQuoteHandler.h
+++ b/programs/odbc-bridge/IdentifierQuoteHandler.h
@ -16,7 +16,7 @@ class IdentifierQuoteHandler : public HTTPRequestHandler, WithContext
 public:
    IdentifierQuoteHandler(size_t keep_alive_timeout_, ContextPtr context_)
        : WithContext(context_)
-        , log(&Poco::Logger::get("IdentifierQuoteHandler"))
+        , log(getLogger("IdentifierQuoteHandler"))
        , keep_alive_timeout(keep_alive_timeout_)
    {
    }
@ -24,7 +24,7 @@ public:
    void handleRequest(HTTPServerRequest & request, HTTPServerResponse & response, const ProfileEvents::Event & write_event) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;
    size_t keep_alive_timeout;
 };

--- a/programs/odbc-bridge/MainHandler.h
+++ b/programs/odbc-bridge/MainHandler.h
@ -24,7 +24,7 @@ public:
        ContextPtr context_,
        const String & mode_)
        : WithContext(context_)
-        , log(&Poco::Logger::get("ODBCHandler"))
+        , log(getLogger("ODBCHandler"))
        , keep_alive_timeout(keep_alive_timeout_)
        , mode(mode_)
    {
@ -33,7 +33,7 @@ public:
    void handleRequest(HTTPServerRequest & request, HTTPServerResponse & response, const ProfileEvents::Event & write_event) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;

    size_t keep_alive_timeout;
    String mode;
--- a/programs/odbc-bridge/ODBCBlockInputStream.cpp
+++ b/programs/odbc-bridge/ODBCBlockInputStream.cpp
@ -23,7 +23,7 @@ namespace ErrorCodes
 ODBCSource::ODBCSource(
    nanodbc::ConnectionHolderPtr connection_holder, const std::string & query_str, const Block & sample_block, const UInt64 max_block_size_)
    : ISource(sample_block)
-    , log(&Poco::Logger::get("ODBCSource"))
+    , log(getLogger("ODBCSource"))
    , max_block_size{max_block_size_}
    , query(query_str)
 {
--- a/programs/odbc-bridge/ODBCBlockInputStream.h
+++ b/programs/odbc-bridge/ODBCBlockInputStream.h
@ -30,7 +30,7 @@ private:
        column.insertFrom(sample_column, 0);
    }

-    Poco::Logger * log;
+    LoggerPtr log;
    const UInt64 max_block_size;
    ExternalResultDescription description;

--- a/programs/odbc-bridge/ODBCBlockOutputStream.cpp
+++ b/programs/odbc-bridge/ODBCBlockOutputStream.cpp
@ -19,7 +19,7 @@ ODBCSink::ODBCSink(
    ContextPtr local_context_,
    IdentifierQuotingStyle quoting_)
    : ISink(sample_block_)
-    , log(&Poco::Logger::get("ODBCSink"))
+    , log(getLogger("ODBCSink"))
    , connection_holder(std::move(connection_holder_))
    , db_name(remote_database_name_)
    , table_name(remote_table_name_)
--- a/programs/odbc-bridge/ODBCBlockOutputStream.h
+++ b/programs/odbc-bridge/ODBCBlockOutputStream.h
@ -30,7 +30,7 @@ protected:
    void consume(Chunk chunk) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;

    nanodbc::ConnectionHolderPtr connection_holder;
    std::string db_name;
--- a/programs/odbc-bridge/ODBCHandlerFactory.cpp
+++ b/programs/odbc-bridge/ODBCHandlerFactory.cpp
@ -11,7 +11,7 @@ namespace DB

 ODBCBridgeHandlerFactory::ODBCBridgeHandlerFactory(const std::string & name_, size_t keep_alive_timeout_, ContextPtr context_)
    : WithContext(context_)
-    , log(&Poco::Logger::get(name_))
+    , log(getLogger(name_))
    , name(name_)
    , keep_alive_timeout(keep_alive_timeout_)
 {
--- a/programs/odbc-bridge/ODBCHandlerFactory.h
+++ b/programs/odbc-bridge/ODBCHandlerFactory.h
@ -22,7 +22,7 @@ public:
    std::unique_ptr<HTTPRequestHandler> createRequestHandler(const HTTPServerRequest & request) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;
    std::string name;
    size_t keep_alive_timeout;
 };
--- a/programs/odbc-bridge/ODBCPooledConnectionFactory.h
+++ b/programs/odbc-bridge/ODBCPooledConnectionFactory.h
@ -97,7 +97,7 @@ T execute(nanodbc::ConnectionHolderPtr connection_holder, std::function<T(nanodb
        /// https://docs.microsoft.com/ru-ru/sql/odbc/reference/appendixes/appendix-a-odbc-error-codes?view=sql-server-ver15
        bool is_retriable = e.state().starts_with("08") || e.state().starts_with("24") || e.state().starts_with("25");
        LOG_ERROR(
-            &Poco::Logger::get("ODBCConnection"),
+            getLogger("ODBCConnection"),
            "ODBC query failed with error: {}, state: {}, native code: {}{}",
            e.what(), e.state(), e.native(), is_retriable ? ", will retry" : "");

--- a/programs/odbc-bridge/SchemaAllowedHandler.h
+++ b/programs/odbc-bridge/SchemaAllowedHandler.h
@ -19,7 +19,7 @@ class SchemaAllowedHandler : public HTTPRequestHandler, WithContext
 public:
    SchemaAllowedHandler(size_t keep_alive_timeout_, ContextPtr context_)
        : WithContext(context_)
-        , log(&Poco::Logger::get("SchemaAllowedHandler"))
+        , log(getLogger("SchemaAllowedHandler"))
        , keep_alive_timeout(keep_alive_timeout_)
    {
    }
@ -27,7 +27,7 @@ public:
    void handleRequest(HTTPServerRequest & request, HTTPServerResponse & response, const ProfileEvents::Event & write_event) override;

 private:
-    Poco::Logger * log;
+    LoggerPtr log;
    size_t keep_alive_timeout;
 };

--- a/programs/odbc-bridge/getIdentifierQuote.cpp
+++ b/programs/odbc-bridge/getIdentifierQuote.cpp
@ -26,7 +26,7 @@ std::string getIdentifierQuote(nanodbc::ConnectionHolderPtr connection_holder)
    }
    catch (...)
    {
-        LOG_WARNING(&Poco::Logger::get("ODBCGetIdentifierQuote"), "Cannot fetch identifier quote. Default double quote is used. Reason: {}", getCurrentExceptionMessage(false));
+        LOG_WARNING(getLogger("ODBCGetIdentifierQuote"), "Cannot fetch identifier quote. Default double quote is used. Reason: {}", getCurrentExceptionMessage(false));
        return "\"";
    }

--- a/programs/server/Server.cpp
+++ b/programs/server/Server.cpp
@ -365,7 +365,7 @@ void Server::createServer(
 namespace
 {

-void setOOMScore(int value, Poco::Logger * log)
+void setOOMScore(int value, LoggerRawPtr log)
 {
    try
    {
@ -450,7 +450,7 @@ void checkForUsersNotInMainConfig(
    const Poco::Util::AbstractConfiguration & config,
    const std::string & config_path,
    const std::string & users_config_path,
-    Poco::Logger * log)
+    LoggerPtr log)
 {
    if (config.getBool("skip_check_for_incorrect_settings", false))
        return;
@ -1740,6 +1740,17 @@ try
    LOG_INFO(log, "Loading metadata from {}", path_str);

    LoadTaskPtrs load_metadata_tasks;
+
+    // Make sure that if exception is thrown during startup async, new async loading jobs are not going to be called.
+    // This is important for the case when exception is thrown from loading of metadata with `async_load_databases = false`
+    // to avoid simultaneously running table startups and destructing databases.
+    SCOPE_EXIT_SAFE(
+        LOG_INFO(log, "Stopping AsyncLoader.");
+
+        // Waits for all currently running jobs to finish and do not run any other pending jobs.
+        global_context->getAsyncLoader().stop();
+    );
+
    try
    {
        auto & database_catalog = DatabaseCatalog::instance();
@ -1888,6 +1899,7 @@ try

        /// Must be done after initialization of `servers`, because async_metrics will access `servers` variable from its thread.
        async_metrics.start();
+        global_context->setAsynchronousMetrics(&async_metrics);

        main_config_reloader->start();
        access_control.startPeriodicReloading();
@ -2478,7 +2490,7 @@ void Server::stopServers(
    const ServerType & server_type
 ) const
 {
-    Poco::Logger * log = &logger();
+    LoggerRawPtr log = &logger();

    /// Remove servers once all their connections are closed
    auto check_server = [&log](const char prefix[], auto & server)
@ -2517,7 +2529,7 @@ void Server::updateServers(
    std::vector<ProtocolServerAdapter> & servers,
    std::vector<ProtocolServerAdapter> & servers_to_start_before_tables)
 {
-    Poco::Logger * log = &logger();
+    LoggerRawPtr log = &logger();

    const auto listen_hosts = getListenHosts(config);
    const auto interserver_listen_hosts = getInterserverListenHosts(config);
--- a/programs/server/binary.html
+++ b/programs/server/binary.html
@ -60,10 +60,16 @@
        /// If it is hosted on server, assume that it is the address of ClickHouse.
        if (location.protocol != 'file:') {
            host = location.origin;
-            user = 'default';
            add_http_cors_header = false;
        }

+        if (window.location.search) {
+            const params = new URLSearchParams(window.location.search);
+            if (params.has('host')) { host = params.get('host'); }
+            if (params.has('user')) { user = params.get('user'); }
+            if (params.has('password')) { password = params.get('password'); }
+        }
+
        let map = L.map('space', {
            crs: L.CRS.Simple,
            center: [-512, 512],
--- a/programs/server/config.xml
+++ b/programs/server/config.xml
@ -1562,6 +1562,10 @@
        <max_entry_size_in_rows>30000000</max_entry_size_in_rows>
    </query_cache>

+    <backups>
+        <allowed_path>backups</allowed_path>
+    </backups>
+
    <!-- This allows to disable exposing addresses in stack traces for security reasons.
         Please be aware that it does not improve security much, but makes debugging much harder.
         The addresses that are small offsets from zero will be displayed nevertheless to show nullptr dereferences.
--- a/programs/server/play.html
+++ b/programs/server/play.html
@ -993,7 +993,16 @@
    function renderError(response)
    {
        clear();
-        document.getElementById('error').innerText = response ? response : "No response.";
+
+        let message = response;
+        try {
+            let json = JSON.parse(response);
+            if (json.exception) {
+                message = json.exception;
+            }
+        } catch (e) {}
+
+        document.getElementById('error').innerText = message ? message : "No response.";
        document.getElementById('error').style.display = 'block';
        document.getElementById('logo-container').style.display = 'none';
    }
--- a/rust/prql/Cargo.toml
+++ b/rust/prql/Cargo.toml
@ -13,7 +13,7 @@ serde_json = "1.0"
 crate-type = ["staticlib"]

 [profile.release]
-debug = true
+debug = false

 [profile.release-thinlto]
 inherits = "release"
--- a/rust/skim/Cargo.toml
+++ b/rust/skim/Cargo.toml
@ -17,7 +17,7 @@ cxx-build = "1.0.83"
 crate-type = ["staticlib"]

 [profile.release]
-debug = true
+debug = false

 [profile.release-thinlto]
 inherits = "release"
--- a/src/Access/AccessControl.cpp
+++ b/src/Access/AccessControl.cpp
@ -47,7 +47,7 @@ namespace
        const Poco::Util::AbstractConfiguration & config,
        const std::string & config_path,
        const std::string & users_config_path,
-        Poco::Logger * log)
+        LoggerPtr log)
    {
        if (config.getBool("skip_check_for_incorrect_settings", false))
            return;
--- a/src/Access/AccessRights.cpp
+++ b/src/Access/AccessRights.cpp
@ -443,7 +443,7 @@ public:
            optimizeTree();
    }

-    void logTree(Poco::Logger * log, const String & title) const
+    void logTree(LoggerPtr log, const String & title) const
    {
        LOG_TRACE(log, "Tree({}): level={}, name={}, flags={}, min_flags={}, max_flags={}, num_children={}",
            title, level, node_name ? *node_name : "NULL", flags.toString(),
@ -1158,7 +1158,7 @@ AccessRights AccessRights::getFullAccess()

 void AccessRights::logTree() const
 {
-    auto * log = &Poco::Logger::get("AccessRights");
+    auto log = getLogger("AccessRights");
    if (root)
    {
        root->logTree(log, "");
--- a/src/Access/Authentication.cpp
+++ b/src/Access/Authentication.cpp
@ -73,7 +73,7 @@ namespace
        return checkPasswordDoubleSHA1MySQL(scramble, scrambled_password, Util::encodeDoubleSHA1(password_plaintext));
    }

-#if USE_SSL
+#if USE_SSH
    bool checkSshSignature(const std::vector<ssh::SSHKey> & keys, std::string_view signature, std::string_view original)
    {
        for (const auto & key: keys)
@ -243,7 +243,7 @@ bool Authentication::areCredentialsValid(
                throw Authentication::Require<SSLCertificateCredentials>("ClickHouse X.509 Authentication");

            case AuthenticationType::SSH_KEY:
-#if USE_SSL
+#if USE_SSH
                return checkSshSignature(auth_data.getSSHKeys(), ssh_credentials->getSignature(), ssh_credentials->getOriginal());
 #else
                throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSH is disabled, because ClickHouse is built without OpenSSL");
--- a/src/Access/AuthenticationData.cpp
+++ b/src/Access/AuthenticationData.cpp
@ -320,7 +320,7 @@ std::shared_ptr<ASTAuthenticationData> AuthenticationData::toAST() const
        }
        case AuthenticationType::SSH_KEY:
        {
-#if USE_SSL
+#if USE_SSH
            for (const auto & key : getSSHKeys())
                node->children.push_back(std::make_shared<ASTPublicSSHKey>(key.getBase64(), key.getKeyType()));

@ -353,7 +353,7 @@ AuthenticationData AuthenticationData::fromAST(const ASTAuthenticationData & que
    /// For this type of authentication we have ASTPublicSSHKey as children for ASTAuthenticationData
    if (query.type && query.type == AuthenticationType::SSH_KEY)
    {
-#if USE_SSL
+#if USE_SSH
        AuthenticationData auth_data(*query.type);
        std::vector<ssh::SSHKey> keys;

--- a/src/Access/Common/AccessType.h
+++ b/src/Access/Common/AccessType.h
@ -170,6 +170,7 @@ enum class AccessType
    M(SYSTEM_RELOAD_MODEL, "SYSTEM RELOAD MODELS, RELOAD MODEL, RELOAD MODELS", GLOBAL, SYSTEM_RELOAD) \
    M(SYSTEM_RELOAD_FUNCTION, "SYSTEM RELOAD FUNCTIONS, RELOAD FUNCTION, RELOAD FUNCTIONS", GLOBAL, SYSTEM_RELOAD) \
    M(SYSTEM_RELOAD_EMBEDDED_DICTIONARIES, "RELOAD EMBEDDED DICTIONARIES", GLOBAL, SYSTEM_RELOAD) /* implicitly enabled by the grant SYSTEM_RELOAD_DICTIONARY ON *.* */\
+    M(SYSTEM_RELOAD_ASYNCHRONOUS_METRICS, "RELOAD ASYNCHRONOUS METRICS", GLOBAL, SYSTEM_RELOAD) \
    M(SYSTEM_RELOAD, "", GROUP, SYSTEM) \
    M(SYSTEM_RESTART_DISK, "SYSTEM RESTART DISK", GLOBAL, SYSTEM) \
    M(SYSTEM_MERGES, "SYSTEM STOP MERGES, SYSTEM START MERGES, STOP MERGES, START MERGES", TABLE, SYSTEM) \
--- a/src/Access/Common/AllowedClientHosts.cpp
+++ b/src/Access/Common/AllowedClientHosts.cpp
@ -514,7 +514,7 @@ bool AllowedClientHosts::contains(const IPAddress & client_address) const
                throw;
            /// Try to ignore DNS errors: if host cannot be resolved, skip it and try next.
            LOG_WARNING(
-                &Poco::Logger::get("AddressPatterns"),
+                getLogger("AddressPatterns"),
                "Failed to check if the allowed client hosts contain address {}. {}, code = {}",
                client_address.toString(), e.displayText(), e.code());
            return false;
@ -556,7 +556,7 @@ bool AllowedClientHosts::contains(const IPAddress & client_address) const
                throw;
            /// Try to ignore DNS errors: if host cannot be resolved, skip it and try next.
            LOG_WARNING(
-                &Poco::Logger::get("AddressPatterns"),
+                getLogger("AddressPatterns"),
                "Failed to check if the allowed client hosts contain address {}. {}, code = {}",
                client_address.toString(), e.displayText(), e.code());
            return false;
--- a/src/Access/ContextAccess.cpp
+++ b/src/Access/ContextAccess.cpp
@ -298,7 +298,7 @@ void ContextAccess::setUser(const UserPtr & user_) const
    }

    user_name = user->getName();
-    trace_log = &Poco::Logger::get("ContextAccess (" + user_name + ")");
+    trace_log = getLogger("ContextAccess (" + user_name + ")");

    std::vector<UUID> current_roles, current_roles_with_admin_option;
    if (params.use_default_roles)
--- a/src/Access/ContextAccess.h
+++ b/src/Access/ContextAccess.h
@ -185,9 +185,10 @@ private:

    mutable std::atomic<bool> initialized = false; // can be removed after Bug 5504 is resolved
    mutable std::atomic<bool> user_was_dropped = false;
-    mutable std::atomic<Poco::Logger *> trace_log = nullptr;

    mutable std::mutex mutex;
+    /// TODO: Fix race
+    mutable LoggerPtr trace_log;
    mutable UserPtr user TSA_GUARDED_BY(mutex);
    mutable String user_name TSA_GUARDED_BY(mutex);
    mutable scope_guard subscription_for_user_change TSA_GUARDED_BY(mutex);
--- a/src/Access/DiskAccessStorage.cpp
+++ b/src/Access/DiskAccessStorage.cpp
@ -47,7 +47,7 @@ namespace
    }


-    AccessEntityPtr tryReadEntityFile(const String & file_path, Poco::Logger & log)
+    AccessEntityPtr tryReadEntityFile(const String & file_path, LoggerPtr log)
    {
        try
        {
@ -55,7 +55,7 @@ namespace
        }
        catch (...)
        {
-            tryLogCurrentException(&log);
+            tryLogCurrentException(log);
            return nullptr;
        }
    }
@ -378,7 +378,7 @@ void DiskAccessStorage::reloadAllAndRebuildLists()
            continue;

        const auto access_entity_file_path = getEntityFilePath(directory_path, id);
-        auto entity = tryReadEntityFile(access_entity_file_path, *getLogger());
+        auto entity = tryReadEntityFile(access_entity_file_path, getLogger());
        if (!entity)
            continue;

--- a/src/Access/ExternalAuthenticators.cpp
+++ b/src/Access/ExternalAuthenticators.cpp
@ -279,7 +279,7 @@ void ExternalAuthenticators::reset()
    resetImpl();
 }

-void ExternalAuthenticators::setConfiguration(const Poco::Util::AbstractConfiguration & config, Poco::Logger * log)
+void ExternalAuthenticators::setConfiguration(const Poco::Util::AbstractConfiguration & config, LoggerPtr log)
 {
    std::lock_guard lock(mutex);
    resetImpl();
--- a/src/Access/ExternalAuthenticators.h
+++ b/src/Access/ExternalAuthenticators.h
@ -36,7 +36,7 @@ class ExternalAuthenticators
 {
 public:
    void reset();
-    void setConfiguration(const Poco::Util::AbstractConfiguration & config, Poco::Logger * log);
+    void setConfiguration(const Poco::Util::AbstractConfiguration & config, LoggerPtr log);

    // The name and readiness of the credentials must be verified before calling these.
    bool checkLDAPCredentials(const String & server, const BasicCredentials & credentials,
--- a/src/Access/GSSAcceptor.cpp
+++ b/src/Access/GSSAcceptor.cpp
@ -328,7 +328,7 @@ void GSSAcceptorContext::initHandles()
    }
 }

-String GSSAcceptorContext::processToken(const String & input_token, Poco::Logger * log)
+String GSSAcceptorContext::processToken(const String & input_token, LoggerPtr log)
 {
    std::lock_guard lock(gss_global_mutex);

@ -455,7 +455,7 @@ void GSSAcceptorContext::initHandles()
 {
 }

-String GSSAcceptorContext::processToken(const String &, Poco::Logger *)
+String GSSAcceptorContext::processToken(const String &, LoggerPtr)
 {
    throw Exception(ErrorCodes::FEATURE_IS_NOT_ENABLED_AT_BUILD_TIME, "ClickHouse was built without GSS-API/Kerberos support");
 }
--- a/src/Access/GSSAcceptor.h
+++ b/src/Access/GSSAcceptor.h
@ -3,6 +3,7 @@
 #include "config.h"

 #include <Access/Credentials.h>
+#include <Common/Logger.h>
 #include <base/types.h>
 #include <memory>

@ -42,7 +43,7 @@ public:

    const String & getRealm() const;
    bool isFailed() const;
-    MAYBE_NORETURN String processToken(const String & input_token, Poco::Logger * log);
+    MAYBE_NORETURN String processToken(const String & input_token, LoggerPtr log);

 private:
    void reset();
--- a/src/Access/IAccessStorage.cpp
+++ b/src/Access/IAccessStorage.cpp
@ -6,6 +6,7 @@
 #include <Backups/BackupEntriesCollector.h>
 #include <Common/Exception.h>
 #include <Common/quoteString.h>
+#include <Common/callOnce.h>
 #include <IO/WriteHelpers.h>
 #include <Interpreters/Context.h>
 #include <Poco/UUIDGenerator.h>
@ -615,7 +616,7 @@ UUID IAccessStorage::generateRandomID()
 }


-void IAccessStorage::clearConflictsInEntitiesList(std::vector<std::pair<UUID, AccessEntityPtr>> & entities, const Poco::Logger * log_)
+void IAccessStorage::clearConflictsInEntitiesList(std::vector<std::pair<UUID, AccessEntityPtr>> & entities, const LoggerPtr log_)
 {
    std::unordered_map<UUID, size_t> positions_by_id;
    std::unordered_map<std::string_view, size_t> positions_by_type_and_name[static_cast<size_t>(AccessEntityType::MAX)];
@ -671,12 +672,13 @@ void IAccessStorage::clearConflictsInEntitiesList(std::vector<std::pair<UUID, Ac
 }


-Poco::Logger * IAccessStorage::getLogger() const
+LoggerPtr IAccessStorage::getLogger() const
 {
-    Poco::Logger * ptr = log.load();
-    if (!ptr)
-        log.store(ptr = &Poco::Logger::get("Access(" + storage_name + ")"), std::memory_order_relaxed);
-    return ptr;
+    callOnce(log_initialized, [&] {
+        log = ::getLogger("Access(" + storage_name + ")");
+    });
+
+    return log;
 }


--- a/src/Access/IAccessStorage.h
+++ b/src/Access/IAccessStorage.h
@ -6,6 +6,7 @@
 #include <Parsers/IParser.h>
 #include <Parsers/parseIdentifierOrStringLiteral.h>
 #include <Common/SettingsChanges.h>
+#include <Common/callOnce.h>

 #include <atomic>
 #include <functional>
@ -225,9 +226,9 @@ protected:
        SettingsChanges & settings) const;
    virtual bool isAddressAllowed(const User & user, const Poco::Net::IPAddress & address) const;
    static UUID generateRandomID();
-    Poco::Logger * getLogger() const;
+    LoggerPtr getLogger() const;
    static String formatEntityTypeWithName(AccessEntityType type, const String & name) { return AccessEntityTypeInfo::get(type).formatEntityNameWithType(name); }
-    static void clearConflictsInEntitiesList(std::vector<std::pair<UUID, AccessEntityPtr>> & entities, const Poco::Logger * log_);
+    static void clearConflictsInEntitiesList(std::vector<std::pair<UUID, AccessEntityPtr>> & entities, const LoggerPtr log_);
    [[noreturn]] void throwNotFound(const UUID & id) const;
    [[noreturn]] void throwNotFound(AccessEntityType type, const String & name) const;
    [[noreturn]] static void throwBadCast(const UUID & id, AccessEntityType type, const String & name, AccessEntityType required_type);
@ -246,7 +247,9 @@ protected:

 private:
    const String storage_name;
-    mutable std::atomic<Poco::Logger *> log = nullptr;
+
+    mutable OnceFlag log_initialized;
+    mutable LoggerPtr log = nullptr;
 };


--- a/src/Access/KerberosInit.cpp
+++ b/src/Access/KerberosInit.cpp
@ -63,7 +63,7 @@ String KerberosInit::fmtError(krb5_error_code code) const

 void KerberosInit::init(const String & keytab_file, const String & principal, const String & cache_name)
 {
-    auto * log = &Poco::Logger::get("KerberosInit");
+    auto log = getLogger("KerberosInit");
    LOG_TRACE(log,"Trying to authenticate with Kerberos v5");

    krb5_error_code ret;
--- a/src/Access/LDAPClient.cpp
+++ b/src/Access/LDAPClient.cpp
@ -532,7 +532,7 @@ LDAPClient::SearchResults LDAPClient::search(const SearchParams & search_params)

                    for (size_t i = 0; referrals[i]; ++i)
                    {
-                        LOG_WARNING(&Poco::Logger::get("LDAPClient"), "Received reference during LDAP search but not following it: {}", referrals[i]);
+                        LOG_WARNING(getLogger("LDAPClient"), "Received reference during LDAP search but not following it: {}", referrals[i]);
                    }
                }

--- a/src/Access/RowPolicyCache.cpp
+++ b/src/Access/RowPolicyCache.cpp
@ -91,7 +91,7 @@ void RowPolicyCache::PolicyInfo::setPolicy(const RowPolicyPtr & policy_)
        catch (...)
        {
            tryLogCurrentException(
-                &Poco::Logger::get("RowPolicy"),
+                getLogger("RowPolicy"),
                String("Could not parse the condition ") + toString(filter_type) + " of row policy "
                    + backQuote(policy->getName()));
        }
--- a/src/Access/SettingsAuthResponseParser.cpp
+++ b/src/Access/SettingsAuthResponseParser.cpp
@ -37,7 +37,7 @@ SettingsAuthResponseParser::parse(const Poco::Net::HTTPResponse & response, std:
    }
    catch (...)
    {
-        LOG_INFO(&Poco::Logger::get("HTTPAuthentication"), "Failed to parse settings from authentication response. Skip it.");
+        LOG_INFO(getLogger("HTTPAuthentication"), "Failed to parse settings from authentication response. Skip it.");
    }
    return result;
 }
--- a/src/Access/UsersConfigAccessStorage.cpp
+++ b/src/Access/UsersConfigAccessStorage.cpp
@ -209,7 +209,7 @@ namespace
        }
        else if (has_ssh_keys)
        {
-#if USE_SSL
+#if USE_SSH
            user->auth_data = AuthenticationData{AuthenticationType::SSH_KEY};

            Poco::Util::AbstractConfiguration::Keys entries;
--- a/src/AggregateFunctions/AggregateFunctionGroupArray.cpp
+++ b/src/AggregateFunctions/AggregateFunctionGroupArray.cpp
@ -440,7 +440,7 @@ struct GroupArrayNodeGeneral : public GroupArrayNodeBase<GroupArrayNodeGeneral>
        return node;
    }

-    void insertInto(IColumn & column) { column.deserializeAndInsertFromArena(data()); }
+    void insertInto(IColumn & column) { std::ignore = column.deserializeAndInsertFromArena(data()); }
 };

 template <typename Node, bool has_sampler>
--- a/src/AggregateFunctions/AggregateFunctionNothing.cpp
+++ b/src/AggregateFunctions/AggregateFunctionNothing.cpp
@ -10,12 +10,28 @@ struct Settings;

 void registerAggregateFunctionNothing(AggregateFunctionFactory & factory)
 {
-    factory.registerFunction("nothing", [](const std::string & name, const DataTypes & argument_types, const Array & parameters, const Settings *)
-    {
-        assertNoParameters(name, parameters);
+    factory.registerFunction(NameAggregateFunctionNothing::name,
+        [](const std::string & name, const DataTypes & argument_types, const Array & parameters, const Settings *)
+        {
+            assertNoParameters(name, parameters);
+            return std::make_shared<AggregateFunctionNothing>(argument_types, parameters);
+        });

-        auto result_type = argument_types.empty() ? std::make_shared<DataTypeNullable>(std::make_shared<DataTypeNothing>()) : argument_types.front();
-        return std::make_shared<AggregateFunctionNothing>(argument_types, parameters, result_type);
+    factory.registerFunction(NameAggregateFunctionNothingNull::name,
+        [](const std::string & name, const DataTypes & argument_types, const Array & parameters, const Settings *)
+        {
+            assertNoParameters(name, parameters);
+            return std::make_shared<AggregateFunctionNothingNull>(argument_types, parameters);
+        });
+
+
+    factory.registerFunction(NameAggregateFunctionNothingUInt64::name, {
+        [](const std::string & name, const DataTypes & argument_types, const Array & parameters, const Settings *)
+        {
+            assertNoParameters(name, parameters);
+            return std::make_shared<AggregateFunctionNothingUInt64>(argument_types, parameters);
+        },
+        AggregateFunctionProperties{ .returns_default_when_only_null = true }
    });
 }

--- a/src/AggregateFunctions/AggregateFunctionNothing.h
+++ b/src/AggregateFunctions/AggregateFunctionNothing.h
@ -6,7 +6,8 @@
 #include <AggregateFunctions/IAggregateFunction.h>
 #include <IO/ReadHelpers.h>
 #include <IO/WriteHelpers.h>
-#include "DataTypes/IDataType.h"
+#include <DataTypes/IDataType.h>
+#include <DataTypes/DataTypesNumber.h>


 namespace DB
@ -18,20 +19,42 @@ namespace ErrorCodes
    extern const int INCORRECT_DATA;
 }

+/// Returns the same type as the first argument
+struct NameAggregateFunctionNothing { static constexpr auto name = "nothing"; };
+/// Always returns Nullable(Nothing)
+struct NameAggregateFunctionNothingNull { static constexpr auto name = "nothingNull"; };
+/// Always returns UInt64
+struct NameAggregateFunctionNothingUInt64 { static constexpr auto name = "nothingUInt64"; };
+
+template <typename Name> class AggregateFunctionNothingImpl;
+
+using AggregateFunctionNothing = AggregateFunctionNothingImpl<NameAggregateFunctionNothing>;
+using AggregateFunctionNothingNull = AggregateFunctionNothingImpl<NameAggregateFunctionNothingNull>;
+using AggregateFunctionNothingUInt64 = AggregateFunctionNothingImpl<NameAggregateFunctionNothingUInt64>;
+

 /** Aggregate function that takes arbitrary number of arbitrary arguments and does nothing.
  */
-class AggregateFunctionNothing final : public IAggregateFunctionHelper<AggregateFunctionNothing>
+template <typename Name>
+class AggregateFunctionNothingImpl final : public IAggregateFunctionHelper<AggregateFunctionNothingImpl<Name>>
 {
-public:
-    AggregateFunctionNothing(const DataTypes & arguments, const Array & params, const DataTypePtr & result_type_)
-        : IAggregateFunctionHelper<AggregateFunctionNothing>(arguments, params, result_type_) {}
-
-    String getName() const override
+    static DataTypePtr getReturnType(const DataTypes & arguments [[maybe_unused]])
    {
-        return "nothing";
+        if constexpr (std::is_same_v<Name, NameAggregateFunctionNothingUInt64>)
+            return std::make_shared<DataTypeUInt64>();
+        else if constexpr (std::is_same_v<Name, NameAggregateFunctionNothingNull>)
+            return std::make_shared<DataTypeNullable>(std::make_shared<DataTypeNothing>());
+        return arguments.empty() ? std::make_shared<DataTypeNullable>(std::make_shared<DataTypeNothing>()) : arguments.front();
    }

+public:
+    AggregateFunctionNothingImpl(const DataTypes & arguments, const Array & params)
+        : IAggregateFunctionHelper<AggregateFunctionNothingImpl<Name>>(arguments, params, getReturnType(arguments))
+    {
+    }
+
+    String getName() const override { return Name::name; }
+
    bool allocatesMemoryInArena() const override { return false; }

    void create(AggregateDataPtr __restrict) const override
@ -75,7 +98,8 @@ public:
        [[maybe_unused]] char symbol;
        readChar(symbol, buf);
        if (symbol != '\0')
-            throw Exception(ErrorCodes::INCORRECT_DATA, "Incorrect state of aggregate function 'nothing', it should contain exactly one zero byte, while it is {}.", static_cast<UInt32>(symbol));
+            throw Exception(ErrorCodes::INCORRECT_DATA, "Incorrect state of aggregate function '{}', it should contain exactly one zero byte, while it is {}",
+                getName(), static_cast<UInt32>(symbol));
    }

    void insertResultInto(AggregateDataPtr __restrict, IColumn & to, Arena *) const override
--- a/src/AggregateFunctions/AggregateFunctionTopK.cpp
+++ b/src/AggregateFunctions/AggregateFunctionTopK.cpp
@ -19,6 +19,7 @@
 #include <Common/assert_cast.h>

 #include <AggregateFunctions/IAggregateFunction.h>
+#include <AggregateFunctions/KeyHolderHelpers.h>


 namespace DB
@ -153,8 +154,6 @@ private:
    UInt64 threshold;
    UInt64 reserved;

-    static void deserializeAndInsert(StringRef str, IColumn & data_to);
-
 public:
    AggregateFunctionTopKGeneric(
        UInt64 threshold_, UInt64 load_factor, const DataTypes & argument_types_, const Array & params)
@ -251,12 +250,7 @@ public:
        offsets_to.push_back(offsets_to.back() + result_vec.size());

        for (auto & elem : result_vec)
-        {
-            if constexpr (is_plain_column)
-                data_to.insertData(elem.key.data, elem.key.size);
-            else
-                data_to.deserializeAndInsertFromArena(elem.key.data);
-        }
+            deserializeAndInsert<is_plain_column>(elem.key, data_to);
    }
 };

--- a/src/AggregateFunctions/Combinators/AggregateFunctionNull.cpp
+++ b/src/AggregateFunctions/Combinators/AggregateFunctionNull.cpp
@ -111,9 +111,9 @@ public:
              * To address this, we handle `nothing` in a special way in `FunctionNode::toASTImpl`.
              */
            if (properties.returns_default_when_only_null)
-                return std::make_shared<AggregateFunctionNothing>(arguments, params, std::make_shared<DataTypeUInt64>());
+                return std::make_shared<AggregateFunctionNothingUInt64>(arguments, params);
            else
-                return std::make_shared<AggregateFunctionNothing>(arguments, params, std::make_shared<DataTypeNullable>(std::make_shared<DataTypeNothing>()));
+                return std::make_shared<AggregateFunctionNothingNull>(arguments, params);
        }

        assert(nested_function);
--- a/src/AggregateFunctions/KeyHolderHelpers.h
+++ b/src/AggregateFunctions/KeyHolderHelpers.h
@ -29,7 +29,7 @@ static void deserializeAndInsert(StringRef str, IColumn & data_to)
    if constexpr (is_plain_column)
        data_to.insertData(str.data, str.size);
    else
-        data_to.deserializeAndInsertFromArena(str.data);
+        std::ignore = data_to.deserializeAndInsertFromArena(str.data);
 }

 }
--- a/src/Analyzer/FunctionNode.cpp
+++ b/src/Analyzer/FunctionNode.cpp
@ -210,18 +210,6 @@ ASTPtr FunctionNode::toASTImpl(const ConvertToASTOptions & options) const
    function_ast->name = function_name;
    function_ast->nulls_action = nulls_action;

-    if (function_name == "nothing")
-    {
-        /** Inside AggregateFunctionCombinatorNull we may replace functions with `NULL` in arguments with `nothing`.
-          * Result type of `nothing` depends on `returns_default_when_only_null` property of nested function.
-          * If we convert `nothing` to AST, we will lose this information, so we use original function name instead.
-          */
-        const auto & original_ast = getOriginalAST();
-        const auto & original_function_ast = original_ast ? original_ast->as<ASTFunction>() : nullptr;
-        if (original_function_ast)
-            function_ast->name = original_function_ast->name;
-    }
-
    if (isWindowFunction())
    {
        function_ast->is_window_function = true;
--- a/src/Analyzer/Passes/FunctionToSubcolumnsPass.cpp
+++ b/src/Analyzer/Passes/FunctionToSubcolumnsPass.cpp
@ -181,6 +181,23 @@ public:

                node = std::make_shared<ColumnNode>(column, column_source);
            }
+            else if (function_name == "variantElement" && isVariant(column_type) && second_argument_constant_node)
+            {
+                /// Replace `variantElement(variant_argument, type_name)` with `variant_argument.type_name`.
+                const auto & variant_element_constant_value = second_argument_constant_node->getValue();
+                String subcolumn_name;
+
+                if (variant_element_constant_value.getType() != Field::Types::String)
+                    return;
+
+                subcolumn_name = variant_element_constant_value.get<const String &>();
+
+                column.name += '.';
+                column.name += subcolumn_name;
+                column.type = function_node->getResultType();
+
+                node = std::make_shared<ColumnNode>(column, column_source);
+            }
            else if (function_name == "mapContains" && column_type.isMap())
            {
                const auto & data_type_map = assert_cast<const DataTypeMap &>(*column.type);
--- a/src/Analyzer/Passes/OptimizeGroupByInjectiveFunctionsPass.cpp
+++ b/src/Analyzer/Passes/OptimizeGroupByInjectiveFunctionsPass.cpp
@ -0,0 +1,124 @@
+#include <Analyzer/Passes/OptimizeGroupByInjectiveFunctionsPass.h>
+#include <Analyzer/ConstantNode.h>
+#include <Analyzer/FunctionNode.h>
+#include <Analyzer/InDepthQueryTreeVisitor.h>
+#include <Analyzer/IQueryTreeNode.h>
+#include <DataTypes/IDataType.h>
+#include <Interpreters/ExternalDictionariesLoader.h>
+
+namespace DB
+{
+
+namespace
+{
+
+const std::unordered_set<String> possibly_injective_function_names
+{
+        "dictGet",
+        "dictGetString",
+        "dictGetUInt8",
+        "dictGetUInt16",
+        "dictGetUInt32",
+        "dictGetUInt64",
+        "dictGetInt8",
+        "dictGetInt16",
+        "dictGetInt32",
+        "dictGetInt64",
+        "dictGetFloat32",
+        "dictGetFloat64",
+        "dictGetDate",
+        "dictGetDateTime"
+};
+
+class OptimizeGroupByInjectiveFunctionsVisitor : public InDepthQueryTreeVisitorWithContext<OptimizeGroupByInjectiveFunctionsVisitor>
+{
+    using Base = InDepthQueryTreeVisitorWithContext<OptimizeGroupByInjectiveFunctionsVisitor>;
+public:
+    explicit OptimizeGroupByInjectiveFunctionsVisitor(ContextPtr context)
+        : Base(std::move(context))
+    {}
+
+    void enterImpl(QueryTreeNodePtr & node)
+    {
+        if (!getSettings().optimize_injective_functions_in_group_by)
+            return;
+
+        auto * query = node->as<QueryNode>();
+        if (!query)
+            return;
+
+        if (!query->hasGroupBy())
+            return;
+
+        if (query->isGroupByWithCube() || query->isGroupByWithRollup())
+            return;
+
+        auto & group_by = query->getGroupBy().getNodes();
+        if (query->isGroupByWithGroupingSets())
+        {
+            for (auto & set : group_by)
+            {
+                auto & grouping_set = set->as<ListNode>()->getNodes();
+                optimizeGroupingSet(grouping_set);
+            }
+        }
+        else
+            optimizeGroupingSet(group_by);
+    }
+
+private:
+    void optimizeGroupingSet(QueryTreeNodes & grouping_set)
+    {
+        auto context = getContext();
+
+        QueryTreeNodes new_group_by_keys;
+        new_group_by_keys.reserve(grouping_set.size());
+        for (auto & group_by_elem : grouping_set)
+        {
+            std::queue<QueryTreeNodePtr> nodes_to_process;
+            nodes_to_process.push(group_by_elem);
+
+            while (!nodes_to_process.empty())
+            {
+                auto node_to_process = nodes_to_process.front();
+                nodes_to_process.pop();
+
+                auto const * function_node = node_to_process->as<FunctionNode>();
+                if (!function_node)
+                {
+                    // Constant aggregation keys are removed in PlannerExpressionAnalysis.cpp
+                    new_group_by_keys.push_back(node_to_process);
+                    continue;
+                }
+
+                // Aggregate functions are not allowed in GROUP BY clause
+                auto function = function_node->getFunctionOrThrow();
+                bool can_be_eliminated = function->isInjective(function_node->getArgumentColumns());
+
+                if (can_be_eliminated)
+                {
+                    for (auto const & argument : function_node->getArguments())
+                    {
+                        // We can skip constants here because aggregation key is already not a constant.
+                        if (argument->getNodeType() != QueryTreeNodeType::CONSTANT)
+                            nodes_to_process.push(argument);
+                    }
+                }
+                else
+                    new_group_by_keys.push_back(node_to_process);
+            }
+        }
+
+        grouping_set = std::move(new_group_by_keys);
+    }
+};
+
+}
+
+void OptimizeGroupByInjectiveFunctionsPass::run(QueryTreeNodePtr query_tree_node, ContextPtr context)
+{
+    OptimizeGroupByInjectiveFunctionsVisitor visitor(std::move(context));
+    visitor.visit(query_tree_node);
+}
+
+}
--- a/src/Analyzer/Passes/OptimizeGroupByInjectiveFunctionsPass.h
+++ b/src/Analyzer/Passes/OptimizeGroupByInjectiveFunctionsPass.h
@ -0,0 +1,20 @@
+#pragma once
+
+#include <Analyzer/IQueryTreePass.h>
+
+namespace DB
+{
+
+/* Eliminates injective functions in GROUP BY section.
+ */
+class OptimizeGroupByInjectiveFunctionsPass final : public IQueryTreePass
+{
+public:
+    String getName() override { return "OptimizeGroupByInjectiveFunctionsPass"; }
+
+    String getDescription() override { return "Replaces injective functions by it's arguments in GROUP BY section."; }
+
+    void run(QueryTreeNodePtr query_tree_node, ContextPtr context) override;
+};
+
+}
--- a/src/Analyzer/Passes/QueryAnalysisPass.cpp
+++ b/src/Analyzer/Passes/QueryAnalysisPass.cpp
@ -2321,11 +2321,15 @@ std::pair<bool, UInt64> QueryAnalyzer::recursivelyCollectMaxOrdinaryExpressions(
  */
 void QueryAnalyzer::expandGroupByAll(QueryNode & query_tree_node_typed)
 {
+    if (!query_tree_node_typed.isGroupByAll())
+        return;
+
    auto & group_by_nodes = query_tree_node_typed.getGroupBy().getNodes();
    auto & projection_list = query_tree_node_typed.getProjection();

    for (auto & node : projection_list.getNodes())
        recursivelyCollectMaxOrdinaryExpressions(node, group_by_nodes);
+    query_tree_node_typed.setIsGroupByAll(false);
 }

 void QueryAnalyzer::expandOrderByAll(QueryNode & query_tree_node_typed)
@ -7422,8 +7426,7 @@ void QueryAnalyzer::resolveQuery(const QueryTreeNodePtr & query_node, Identifier
        node->removeAlias();
    }

-    if (query_node_typed.isGroupByAll())
-        expandGroupByAll(query_node_typed);
+    expandGroupByAll(query_node_typed);

    validateFilters(query_node);
    validateAggregates(query_node, { .group_by_use_nulls = scope.group_by_use_nulls });
--- a/src/Analyzer/Passes/RewriteSumFunctionWithSumAndCountPass.cpp
+++ b/src/Analyzer/Passes/RewriteSumFunctionWithSumAndCountPass.cpp
@ -10,14 +10,6 @@
 namespace DB
 {

-/** Rewrites `sum(column +/- literal)` into two individual functions
- * `sum(column)` and `literal * count(column)`.
- * sum(column + literal) -> sum(column) + literal * count(column)
- * sum(literal + column) -> literal * count(column) + sum(column)
- * sum(column - literal) -> sum(column) - literal * count(column)
- * sum(literal - column) -> literal * count(column) - sum(column)
- */
-
 namespace
 {

@ -29,6 +21,9 @@ public:

    void enterImpl(QueryTreeNodePtr & node)
    {
+        if (!getSettings().optimize_arithmetic_operations_in_aggregate_functions)
+            return;
+
        static const std::unordered_set<String> func_supported = {
            "plus",
            "minus"
--- a/src/Analyzer/Passes/RewriteSumFunctionWithSumAndCountPass.h
+++ b/src/Analyzer/Passes/RewriteSumFunctionWithSumAndCountPass.h
@ -5,6 +5,14 @@
 namespace DB
 {

+/**
+ * Rewrites `sum(column +/- literal)` into two individual functions
+ * `sum(column)` and `literal * count(column)`.
+ * sum(column + literal) -> sum(column) + literal * count(column)
+ * sum(literal + column) -> literal * count(column) + sum(column)
+ * sum(column - literal) -> sum(column) - literal * count(column)
+ * sum(literal - column) -> literal * count(column) - sum(column)
+ */
 class RewriteSumFunctionWithSumAndCountPass final : public IQueryTreePass
 {
 public:
--- a/src/Analyzer/QueryTreePassManager.cpp
+++ b/src/Analyzer/QueryTreePassManager.cpp
@ -3,6 +3,7 @@
 #include <memory>

 #include <Common/Exception.h>
+#include "Analyzer/Passes/OptimizeGroupByInjectiveFunctionsPass.h"

 #include <IO/WriteHelpers.h>
 #include <IO/Operators.h>
@ -163,8 +164,6 @@ private:

 /** ClickHouse query tree pass manager.
  *
-  * TODO: Support setting optimize_substitute_columns.
-  * TODO: Support GROUP BY injective function elimination.
  * TODO: Support setting optimize_aggregators_of_group_by_keys.
  * TODO: Support setting optimize_monotonous_functions_in_order_by.
  * TODO: Add optimizations based on function semantics. Example: SELECT * FROM test_table WHERE id != id. (id is not nullable column).
@ -268,6 +267,7 @@ void addQueryTreePasses(QueryTreePassManager & manager)
    manager.addPass(std::make_unique<AggregateFunctionsArithmericOperationsPass>());
    manager.addPass(std::make_unique<UniqInjectiveFunctionsEliminationPass>());
    manager.addPass(std::make_unique<OptimizeGroupByFunctionKeysPass>());
+    manager.addPass(std::make_unique<OptimizeGroupByInjectiveFunctionsPass>());

    manager.addPass(std::make_unique<MultiIfToIfPass>());
    manager.addPass(std::make_unique<IfConstantConditionPass>());
--- a/src/Analyzer/Utils.cpp
+++ b/src/Analyzer/Utils.cpp
@ -326,6 +326,68 @@ void addTableExpressionOrJoinIntoTablesInSelectQuery(ASTPtr & tables_in_select_q
    }
 }

+QueryTreeNodes extractAllTableReferences(const QueryTreeNodePtr & tree)
+{
+    QueryTreeNodes result;
+
+    QueryTreeNodes nodes_to_process;
+    nodes_to_process.push_back(tree);
+
+    while (!nodes_to_process.empty())
+    {
+        auto node_to_process = std::move(nodes_to_process.back());
+        nodes_to_process.pop_back();
+
+        auto node_type = node_to_process->getNodeType();
+
+        switch (node_type)
+        {
+            case QueryTreeNodeType::TABLE:
+            {
+                result.push_back(std::move(node_to_process));
+                break;
+            }
+            case QueryTreeNodeType::QUERY:
+            {
+                nodes_to_process.push_back(node_to_process->as<QueryNode>()->getJoinTree());
+                break;
+            }
+            case QueryTreeNodeType::UNION:
+            {
+                for (const auto & union_node : node_to_process->as<UnionNode>()->getQueries().getNodes())
+                    nodes_to_process.push_back(union_node);
+                break;
+            }
+            case QueryTreeNodeType::TABLE_FUNCTION:
+            {
+                // Arguments of table function can't contain TableNodes.
+                break;
+            }
+            case QueryTreeNodeType::ARRAY_JOIN:
+            {
+                nodes_to_process.push_back(node_to_process->as<ArrayJoinNode>()->getTableExpression());
+                break;
+            }
+            case QueryTreeNodeType::JOIN:
+            {
+                auto & join_node = node_to_process->as<JoinNode &>();
+                nodes_to_process.push_back(join_node.getRightTableExpression());
+                nodes_to_process.push_back(join_node.getLeftTableExpression());
+                break;
+            }
+            default:
+            {
+                throw Exception(ErrorCodes::LOGICAL_ERROR,
+                                "Unexpected node type for table expression. "
+                                "Expected table, table function, query, union, join or array join. Actual {}",
+                                node_to_process->getNodeTypeName());
+            }
+        }
+    }
+
+    return result;
+}
+
 QueryTreeNodes extractTableExpressions(const QueryTreeNodePtr & join_tree_node, bool add_array_join)
 {
    QueryTreeNodes result;
@ -626,7 +688,7 @@ void rerunFunctionResolve(FunctionNode * function_node, ContextPtr context)
    }
    else if (function_node->isAggregateFunction())
    {
-        if (name == "nothing")
+        if (name == "nothing" || name == "nothingUInt64" || name == "nothingNull")
            return;
        function_node->resolveAsAggregateFunction(resolveAggregateFunction(function_node));
    }
--- a/src/Analyzer/Utils.h
+++ b/src/Analyzer/Utils.h
@ -50,10 +50,13 @@ std::optional<bool> tryExtractConstantFromConditionNode(const QueryTreeNodePtr &
  */
 void addTableExpressionOrJoinIntoTablesInSelectQuery(ASTPtr & tables_in_select_query_ast, const QueryTreeNodePtr & table_expression, const IQueryTreeNode::ConvertToASTOptions & convert_to_ast_options);

-/// Extract table, table function, query, union from join tree
+/// Extract all TableNodes from the query tree.
+QueryTreeNodes extractAllTableReferences(const QueryTreeNodePtr & tree);
+
+/// Extract table, table function, query, union from join tree.
 QueryTreeNodes extractTableExpressions(const QueryTreeNodePtr & join_tree_node, bool add_array_join = false);

-/// Extract left table expression from join tree
+/// Extract left table expression from join tree.
 QueryTreeNodePtr extractLeftTableExpression(const QueryTreeNodePtr & join_tree_node);

 /** Build table expressions stack that consists from table, table function, query, union, join, array join from join tree.
--- a/src/Backups/BackupCoordinationLocal.cpp
+++ b/src/Backups/BackupCoordinationLocal.cpp
@ -9,7 +9,7 @@ namespace DB
 {

 BackupCoordinationLocal::BackupCoordinationLocal(bool plain_backup_)
-    : log(&Poco::Logger::get("BackupCoordinationLocal")), file_infos(plain_backup_)
+    : log(getLogger("BackupCoordinationLocal")), file_infos(plain_backup_)
 {
 }

--- a/src/Backups/BackupCoordinationLocal.h
+++ b/src/Backups/BackupCoordinationLocal.h
@ -57,7 +57,7 @@ public:
    bool hasConcurrentBackups(const std::atomic<size_t> & num_active_backups) const override;

 private:
-    Poco::Logger * const log;
+    LoggerPtr const log;

    BackupCoordinationReplicatedTables TSA_GUARDED_BY(replicated_tables_mutex) replicated_tables;
    BackupCoordinationReplicatedAccess TSA_GUARDED_BY(replicated_access_mutex) replicated_access;
--- a/src/Backups/BackupCoordinationRemote.cpp
+++ b/src/Backups/BackupCoordinationRemote.cpp
@ -173,7 +173,7 @@ BackupCoordinationRemote::BackupCoordinationRemote(
    , current_host_index(findCurrentHostIndex(all_hosts, current_host))
    , plain_backup(plain_backup_)
    , is_internal(is_internal_)
-    , log(&Poco::Logger::get("BackupCoordinationRemote"))
+    , log(getLogger("BackupCoordinationRemote"))
    , with_retries(
        log,
        get_zookeeper_,
--- a/src/Backups/BackupCoordinationRemote.h
+++ b/src/Backups/BackupCoordinationRemote.h
@ -102,7 +102,7 @@ private:
    const size_t current_host_index;
    const bool plain_backup;
    const bool is_internal;
-    Poco::Logger * const log;
+    LoggerPtr const log;

    /// The order of these two fields matters, because stage_sync holds a reference to with_retries object
    mutable WithRetries with_retries;
--- a/Show More
+++ b/Show More