Merge branch 'master' into adaptive_parallel_replicas

2024-09-19 16:20:50 +00:00 · 2024-08-19 12:48:39 +02:00 · 2024-08-19 12:48:39 +02:00 · d4a3a033b0
commit d4a3a033b0
parent c252b3c8b0 46dd66bfb6
174 changed files with 833 additions and 512 deletions
--- a/contrib/aws
+++ b/contrib/aws
@ -1 +1 @@
-Subproject commit 1c2946bfcb7f1e3ae0a858de0b59d4f1a7b4ccaf
+Subproject commit d5450d76abda556ce145ddabe7e0cc6a7644ec59
--- a/contrib/aws-crt-cpp
+++ b/contrib/aws-crt-cpp
@ -1 +1 @@
-Subproject commit f532d6abc0d2b0d8b5d6fe9e7c51eaedbe4afbd0
+Subproject commit e5aa45cacfdcda7719ead38760e7c61076f5745f
--- a/docker/test/stateful/run.sh
+++ b/docker/test/stateful/run.sh
@ -118,8 +118,8 @@ if [[ -n "$USE_DATABASE_REPLICATED" ]] && [[ "$USE_DATABASE_REPLICATED" -eq 1 ]]
    clickhouse-client --query "CREATE TABLE test.hits AS datasets.hits_v1"
    clickhouse-client --query "CREATE TABLE test.visits AS datasets.visits_v1"

-    clickhouse-client --query "INSERT INTO test.hits SELECT * FROM datasets.hits_v1"
-    clickhouse-client --query "INSERT INTO test.visits SELECT * FROM datasets.visits_v1"
+    clickhouse-client --max_memory_usage 10G --query "INSERT INTO test.hits SELECT * FROM datasets.hits_v1"
+    clickhouse-client --max_memory_usage 10G --query "INSERT INTO test.visits SELECT * FROM datasets.visits_v1"

    clickhouse-client --query "DROP TABLE datasets.hits_v1"
    clickhouse-client --query "DROP TABLE datasets.visits_v1"
@ -191,8 +191,8 @@ else
            ENGINE = CollapsingMergeTree(Sign) PARTITION BY toYYYYMM(StartDate) ORDER BY (CounterID, StartDate, intHash32(UserID), VisitID)
            SAMPLE BY intHash32(UserID) SETTINGS index_granularity = 8192, storage_policy='s3_cache'"

-        clickhouse-client --query "INSERT INTO test.hits SELECT * FROM datasets.hits_v1 SETTINGS enable_filesystem_cache_on_write_operations=0, max_insert_threads=16"
-        clickhouse-client --query "INSERT INTO test.visits SELECT * FROM datasets.visits_v1 SETTINGS enable_filesystem_cache_on_write_operations=0, max_insert_threads=16"
+        clickhouse-client --max_memory_usage 10G --query "INSERT INTO test.hits SELECT * FROM datasets.hits_v1 SETTINGS enable_filesystem_cache_on_write_operations=0, max_insert_threads=16"
+        clickhouse-client --max_memory_usage 10G --query "INSERT INTO test.visits SELECT * FROM datasets.visits_v1 SETTINGS enable_filesystem_cache_on_write_operations=0, max_insert_threads=16"
        clickhouse-client --query "DROP TABLE datasets.visits_v1 SYNC"
        clickhouse-client --query "DROP TABLE datasets.hits_v1 SYNC"
    else
@ -200,7 +200,8 @@ else
        clickhouse-client --query "RENAME TABLE datasets.visits_v1 TO test.visits"
    fi
    clickhouse-client --query "CREATE TABLE test.hits_s3  (WatchID UInt64, JavaEnable UInt8, Title String, GoodEvent Int16, EventTime DateTime, EventDate Date, CounterID UInt32, ClientIP UInt32, ClientIP6 FixedString(16), RegionID UInt32, UserID UInt64, CounterClass Int8, OS UInt8, UserAgent UInt8, URL String, Referer String, URLDomain String, RefererDomain String, Refresh UInt8, IsRobot UInt8, RefererCategories Array(UInt16), URLCategories Array(UInt16), URLRegions Array(UInt32), RefererRegions Array(UInt32), ResolutionWidth UInt16, ResolutionHeight UInt16, ResolutionDepth UInt8, FlashMajor UInt8, FlashMinor UInt8, FlashMinor2 String, NetMajor UInt8, NetMinor UInt8, UserAgentMajor UInt16, UserAgentMinor FixedString(2), CookieEnable UInt8, JavascriptEnable UInt8, IsMobile UInt8, MobilePhone UInt8, MobilePhoneModel String, Params String, IPNetworkID UInt32, TraficSourceID Int8, SearchEngineID UInt16, SearchPhrase String, AdvEngineID UInt8, IsArtifical UInt8, WindowClientWidth UInt16, WindowClientHeight UInt16, ClientTimeZone Int16, ClientEventTime DateTime, SilverlightVersion1 UInt8, SilverlightVersion2 UInt8, SilverlightVersion3 UInt32, SilverlightVersion4 UInt16, PageCharset String, CodeVersion UInt32, IsLink UInt8, IsDownload UInt8, IsNotBounce UInt8, FUniqID UInt64, HID UInt32, IsOldCounter UInt8, IsEvent UInt8, IsParameter UInt8, DontCountHits UInt8, WithHash UInt8, HitColor FixedString(1), UTCEventTime DateTime, Age UInt8, Sex UInt8, Income UInt8, Interests UInt16, Robotness UInt8, GeneralInterests Array(UInt16), RemoteIP UInt32, RemoteIP6 FixedString(16), WindowName Int32, OpenerName Int32, HistoryLength Int16, BrowserLanguage FixedString(2), BrowserCountry FixedString(2), SocialNetwork String, SocialAction String, HTTPError UInt16, SendTiming Int32, DNSTiming Int32, ConnectTiming Int32, ResponseStartTiming Int32, ResponseEndTiming Int32, FetchTiming Int32, RedirectTiming Int32, DOMInteractiveTiming Int32, DOMContentLoadedTiming Int32, DOMCompleteTiming Int32, LoadEventStartTiming Int32, LoadEventEndTiming Int32, NSToDOMContentLoadedTiming Int32, FirstPaintTiming Int32, RedirectCount Int8, SocialSourceNetworkID UInt8, SocialSourcePage String, ParamPrice Int64, ParamOrderID String, ParamCurrency FixedString(3), ParamCurrencyID UInt16, GoalsReached Array(UInt32), OpenstatServiceName String, OpenstatCampaignID String, OpenstatAdID String, OpenstatSourceID String, UTMSource String, UTMMedium String, UTMCampaign String, UTMContent String, UTMTerm String, FromTag String, HasGCLID UInt8, RefererHash UInt64, URLHash UInt64, CLID UInt32, YCLID UInt64, ShareService String, ShareURL String, ShareTitle String, ParsedParams Nested(Key1 String, Key2 String, Key3 String, Key4 String, Key5 String, ValueDouble Float64), IslandID FixedString(16), RequestNum UInt32, RequestTry UInt8) ENGINE = MergeTree() PARTITION BY toYYYYMM(EventDate) ORDER BY (CounterID, EventDate, intHash32(UserID)) SAMPLE BY intHash32(UserID) SETTINGS index_granularity = 8192, storage_policy='s3_cache'"
-    clickhouse-client --query "INSERT INTO test.hits_s3 SELECT * FROM test.hits SETTINGS enable_filesystem_cache_on_write_operations=0, max_insert_threads=16"
+    # AWS S3 is very inefficient, so increase memory even further:
+    clickhouse-client --max_memory_usage 30G --max_memory_usage_for_user 30G --query "INSERT INTO test.hits_s3 SELECT * FROM test.hits SETTINGS enable_filesystem_cache_on_write_operations=0, max_insert_threads=16"
 fi

 clickhouse-client --query "SHOW TABLES FROM test"
--- a/docker/test/stateless/run.sh
+++ b/docker/test/stateless/run.sh
@ -391,8 +391,8 @@ done
 # wait for minio to flush its batch if it has any
 sleep 1
 clickhouse-client -q "SYSTEM FLUSH ASYNC INSERT QUEUE"
-clickhouse-client -q "SELECT log FROM minio_audit_logs ORDER BY event_time INTO OUTFILE '/test_output/minio_audit_logs.jsonl.zst' FORMAT JSONEachRow"
-clickhouse-client -q "SELECT log FROM minio_server_logs ORDER BY event_time INTO OUTFILE '/test_output/minio_server_logs.jsonl.zst' FORMAT JSONEachRow"
+clickhouse-client --max_block_size 8192 --max_memory_usage 10G --max_threads 1 --max_result_bytes 0 --max_result_rows 0 --max_rows_to_read 0 --max_bytes_to_read 0 -q "SELECT log FROM minio_audit_logs ORDER BY event_time INTO OUTFILE '/test_output/minio_audit_logs.jsonl.zst' FORMAT JSONEachRow"
+clickhouse-client --max_block_size 8192 --max_memory_usage 10G --max_threads 1 --max_result_bytes 0 --max_result_rows 0 --max_rows_to_read 0 --max_bytes_to_read 0 -q "SELECT log FROM minio_server_logs ORDER BY event_time INTO OUTFILE '/test_output/minio_server_logs.jsonl.zst' FORMAT JSONEachRow"

 # Stop server so we can safely read data with clickhouse-local.
 # Why do we read data with clickhouse-local?
--- a/programs/client/Client.cpp
+++ b/programs/client/Client.cpp
@ -1164,6 +1164,9 @@ void Client::processOptions(const OptionsDescription & options_description,
    /// (There is no need to copy the context because clickhouse-client has no background tasks so it won't use that context in parallel.)
    client_context = global_context;
    initClientContext();
+
+    /// Allow to pass-through unknown settings to the server.
+    client_context->getAccessControl().allowAllSettings();
 }


--- a/programs/format/Format.cpp
+++ b/programs/format/Format.cpp
@ -264,7 +264,11 @@ int mainEntryClickHouseFormat(int argc, char ** argv)
                    if (!backslash)
                    {
                        WriteBufferFromOwnString str_buf;
-                        formatAST(*res, str_buf, hilite, oneline || approx_query_length < max_line_length);
+                        bool oneline_current_query = oneline || approx_query_length < max_line_length;
+                        IAST::FormatSettings settings(str_buf, oneline_current_query, hilite);
+                        settings.show_secrets = true;
+                        settings.print_pretty_type_names = !oneline_current_query;
+                        res->format(settings);

                        if (insert_query_payload)
                        {
@ -307,7 +311,11 @@ int mainEntryClickHouseFormat(int argc, char ** argv)
                    else
                    {
                        WriteBufferFromOwnString str_buf;
-                        formatAST(*res, str_buf, hilite, oneline);
+                        bool oneline_current_query = oneline || approx_query_length < max_line_length;
+                        IAST::FormatSettings settings(str_buf, oneline_current_query, hilite);
+                        settings.show_secrets = true;
+                        settings.print_pretty_type_names = !oneline_current_query;
+                        res->format(settings);

                        auto res_string = str_buf.str();
                        WriteBufferFromOStream res_cout(std::cout, 4096);
--- a/programs/keeper/Keeper.cpp
+++ b/programs/keeper/Keeper.cpp
@ -66,6 +66,8 @@
 /// A minimal file used when the keeper is run without installation
 INCBIN(keeper_resource_embedded_xml, SOURCE_DIR "/programs/keeper/keeper_embedded.xml");

+extern const char * GIT_HASH;
+
 int mainEntryClickHouseKeeper(int argc, char ** argv)
 {
    DB::Keeper app;
@ -675,7 +677,7 @@ void Keeper::logRevision() const
        "Starting ClickHouse Keeper {} (revision: {}, git hash: {}, build id: {}), PID {}",
        VERSION_STRING,
        ClickHouseRevision::getVersionRevision(),
-        git_hash.empty() ? "<unknown>" : git_hash,
+        GIT_HASH,
        build_id.empty() ? "<unknown>" : build_id,
        getpid());
 }
--- a/programs/local/LocalServer.cpp
+++ b/programs/local/LocalServer.cpp
@ -367,7 +367,7 @@ std::string LocalServer::getInitialCreateTableQuery()
    else
        table_structure = "(" + table_structure + ")";

-    return fmt::format("CREATE TABLE {} {} ENGINE = File({}, {});",
+    return fmt::format("CREATE TEMPORARY TABLE {} {} ENGINE = File({}, {});",
                       table_name, table_structure, data_format, table_file);
 }

--- a/programs/server/Server.cpp
+++ b/programs/server/Server.cpp
@ -1920,7 +1920,7 @@ try
    auto & access_control = global_context->getAccessControl();
    try
    {
-        access_control.setUpFromMainConfig(config(), config_path, [&] { return global_context->getZooKeeper(); });
+        access_control.setupFromMainConfig(config(), config_path, [&] { return global_context->getZooKeeper(); });
    }
    catch (...)
    {
--- a/src/Access/AccessControl.cpp
+++ b/src/Access/AccessControl.cpp
@ -280,7 +280,7 @@ void AccessControl::shutdown()
 }


-void AccessControl::setUpFromMainConfig(const Poco::Util::AbstractConfiguration & config_, const String & config_path_,
+void AccessControl::setupFromMainConfig(const Poco::Util::AbstractConfiguration & config_, const String & config_path_,
                                        const zkutil::GetZooKeeper & get_zookeeper_function_)
 {
    if (config_.has("custom_settings_prefixes"))
@ -868,4 +868,10 @@ const ExternalAuthenticators & AccessControl::getExternalAuthenticators() const
    return *external_authenticators;
 }

+
+void AccessControl::allowAllSettings()
+{
+    custom_settings_prefixes->registerPrefixes({""});
+}
+
 }
--- a/src/Access/AccessControl.h
+++ b/src/Access/AccessControl.h
@ -57,7 +57,7 @@ public:
    void shutdown() override;

    /// Initializes access storage (user directories).
-    void setUpFromMainConfig(const Poco::Util::AbstractConfiguration & config_, const String & config_path_,
+    void setupFromMainConfig(const Poco::Util::AbstractConfiguration & config_, const String & config_path_,
                             const zkutil::GetZooKeeper & get_zookeeper_function_);

    /// Parses access entities from a configuration loaded from users.xml.
@ -238,6 +238,9 @@ public:
    /// Gets manager of notifications.
    AccessChangesNotifier & getChangesNotifier();

+    /// Allow all setting names - this can be used in clients to pass-through unknown settings to the server.
+    void allowAllSettings();
+
 private:
    class ContextAccessCache;
    class CustomSettingsPrefixes;
--- a/src/Access/SettingsConstraints.cpp
+++ b/src/Access/SettingsConstraints.cpp
@ -219,8 +219,8 @@ void SettingsConstraints::clamp(const Settings & current_settings, SettingsChang
        });
 }

-template <class T>
-bool getNewValueToCheck(const T & current_settings, SettingChange & change, Field & new_value, bool throw_on_failure)
+template <typename SettingsT>
+bool getNewValueToCheck(const SettingsT & current_settings, SettingChange & change, Field & new_value, bool throw_on_failure)
 {
    Field current_value;
    bool has_current_value = current_settings.tryGet(change.name, current_value);
@ -230,12 +230,12 @@ bool getNewValueToCheck(const T & current_settings, SettingChange & change, Fiel
        return false;

    if (throw_on_failure)
-        new_value = T::castValueUtil(change.name, change.value);
+        new_value = SettingsT::castValueUtil(change.name, change.value);
    else
    {
        try
        {
-            new_value = T::castValueUtil(change.name, change.value);
+            new_value = SettingsT::castValueUtil(change.name, change.value);
        }
        catch (...)
        {
--- a/src/Client/ClientBase.cpp
+++ b/src/Client/ClientBase.cpp
@ -58,6 +58,7 @@
 #include <QueryPipeline/QueryPipelineBuilder.h>
 #include <Interpreters/ReplaceQueryParameterVisitor.h>
 #include <Interpreters/ProfileEventsExt.h>
+#include <Interpreters/InterpreterSetQuery.h>
 #include <IO/WriteBufferFromOStream.h>
 #include <IO/WriteBufferFromFileDescriptor.h>
 #include <IO/CompressionMethod.h>
@ -331,7 +332,11 @@ ASTPtr ClientBase::parseQuery(const char *& pos, const char * end, const Setting
    {
        output_stream << std::endl;
        WriteBufferFromOStream res_buf(output_stream, 4096);
-        formatAST(*res, res_buf);
+        IAST::FormatSettings format_settings(res_buf, /* one_line */ false);
+        format_settings.hilite = true;
+        format_settings.show_secrets = true;
+        format_settings.print_pretty_type_names = true;
+        res->format(format_settings);
        res_buf.finalize();
        output_stream << std::endl << std::endl;
    }
@ -1604,14 +1609,14 @@ void ClientBase::sendData(Block & sample, const ColumnsDescription & columns_des
            auto metadata = storage->getInMemoryMetadataPtr();
            QueryPlan plan;
            storage->read(
-                    plan,
-                    sample.getNames(),
-                    storage->getStorageSnapshot(metadata, client_context),
-                    query_info,
-                    client_context,
-                    {},
-                    client_context->getSettingsRef().max_block_size,
-                    getNumberOfPhysicalCPUCores());
+                plan,
+                sample.getNames(),
+                storage->getStorageSnapshot(metadata, client_context),
+                query_info,
+                client_context,
+                {},
+                client_context->getSettingsRef().max_block_size,
+                getNumberOfPhysicalCPUCores());

            auto builder = plan.buildQueryPipeline(
                QueryPlanOptimizationSettings::fromContext(client_context),
@ -1888,48 +1893,19 @@ void ClientBase::processParsedSingleQuery(const String & full_query, const Strin
    profile_events.watch.restart();

    {
-        /// Temporarily apply query settings to context.
-        std::optional<Settings> old_settings;
-        SCOPE_EXIT_SAFE({
-            if (old_settings)
-                client_context->setSettings(*old_settings);
+        /// Temporarily apply query settings to the context.
+        Settings old_settings = client_context->getSettingsCopy();
+        SCOPE_EXIT_SAFE(
+        {
+            client_context->setSettings(old_settings);
        });
-
-        auto apply_query_settings = [&](const IAST & settings_ast)
-        {
-            if (!old_settings)
-                old_settings.emplace(client_context->getSettingsRef());
-            client_context->applySettingsChanges(settings_ast.as<ASTSetQuery>()->changes);
-            client_context->resetSettingsToDefaultValue(settings_ast.as<ASTSetQuery>()->default_settings);
-        };
-
-        const auto * insert = parsed_query->as<ASTInsertQuery>();
-        if (const auto * select = parsed_query->as<ASTSelectQuery>(); select && select->settings())
-            apply_query_settings(*select->settings());
-        else if (const auto * select_with_union = parsed_query->as<ASTSelectWithUnionQuery>())
-        {
-            const ASTs & children = select_with_union->list_of_selects->children;
-            if (!children.empty())
-            {
-                // On the client it is enough to apply settings only for the
-                // last SELECT, since the only thing that is important to apply
-                // on the client is format settings.
-                const auto * last_select = children.back()->as<ASTSelectQuery>();
-                if (last_select && last_select->settings())
-                {
-                    apply_query_settings(*last_select->settings());
-                }
-            }
-        }
-        else if (const auto * query_with_output = parsed_query->as<ASTQueryWithOutput>(); query_with_output && query_with_output->settings_ast)
-            apply_query_settings(*query_with_output->settings_ast);
-        else if (insert && insert->settings_ast)
-            apply_query_settings(*insert->settings_ast);
+        InterpreterSetQuery::applySettingsFromQuery(parsed_query, client_context);

        if (!connection->checkConnected(connection_parameters.timeouts))
            connect();

        ASTPtr input_function;
+        const auto * insert = parsed_query->as<ASTInsertQuery>();
        if (insert && insert->select)
            insert->tryFindInputFunction(input_function);

--- a/src/Common/ProxyConfigurationResolverProvider.cpp
+++ b/src/Common/ProxyConfigurationResolverProvider.cpp
@ -112,9 +112,8 @@ namespace
        return configuration.has(config_prefix + ".uri");
    }

-    /*
-     * New syntax requires protocol prefix "<http> or <https>"
-     * */
+    /* New syntax requires protocol prefix "<http> or <https>"
+     */
    std::optional<std::string> getProtocolPrefix(
        ProxyConfiguration::Protocol request_protocol,
        const String & config_prefix,
@ -130,22 +129,18 @@ namespace
        return protocol_prefix;
    }

-    template <bool new_syntax>
    std::optional<std::string> calculatePrefixBasedOnSettingsSyntax(
+        bool new_syntax,
        ProxyConfiguration::Protocol request_protocol,
        const String & config_prefix,
        const Poco::Util::AbstractConfiguration & configuration
    )
    {
        if (!configuration.has(config_prefix))
-        {
            return std::nullopt;
-        }

-        if constexpr (new_syntax)
-        {
+        if (new_syntax)
            return getProtocolPrefix(request_protocol, config_prefix, configuration);
-        }

        return config_prefix;
    }
@ -155,24 +150,21 @@ std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::
    Protocol request_protocol,
    const Poco::Util::AbstractConfiguration & configuration)
 {
-    if (auto resolver = getFromSettings(request_protocol, "proxy", configuration))
-    {
+    if (auto resolver = getFromSettings(true, request_protocol, "proxy", configuration))
        return resolver;
-    }

    return std::make_shared<EnvironmentProxyConfigurationResolver>(
        request_protocol,
        isTunnelingDisabledForHTTPSRequestsOverHTTPProxy(configuration));
 }

-template <bool is_new_syntax>
 std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::getFromSettings(
+    bool new_syntax,
    Protocol request_protocol,
    const String & config_prefix,
-    const Poco::Util::AbstractConfiguration & configuration
-)
+    const Poco::Util::AbstractConfiguration & configuration)
 {
-    auto prefix_opt = calculatePrefixBasedOnSettingsSyntax<is_new_syntax>(request_protocol, config_prefix, configuration);
+    auto prefix_opt = calculatePrefixBasedOnSettingsSyntax(new_syntax, request_protocol, config_prefix, configuration);

    if (!prefix_opt)
    {
@ -195,20 +187,17 @@ std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::
 std::shared_ptr<ProxyConfigurationResolver> ProxyConfigurationResolverProvider::getFromOldSettingsFormat(
    Protocol request_protocol,
    const String & config_prefix,
-    const Poco::Util::AbstractConfiguration & configuration
-)
+    const Poco::Util::AbstractConfiguration & configuration)
 {
-    /*
-     * First try to get it from settings only using the combination of config_prefix and configuration.
+    /* First try to get it from settings only using the combination of config_prefix and configuration.
     * This logic exists for backward compatibility with old S3 storage specific proxy configuration.
     * */
-    if (auto resolver = ProxyConfigurationResolverProvider::getFromSettings<false>(request_protocol, config_prefix + ".proxy", configuration))
+    if (auto resolver = ProxyConfigurationResolverProvider::getFromSettings(false, request_protocol, config_prefix + ".proxy", configuration))
    {
        return resolver;
    }

-    /*
-     * In case the combination of config_prefix and configuration does not provide a resolver, try to get it from general / new settings.
+    /* In case the combination of config_prefix and configuration does not provide a resolver, try to get it from general / new settings.
     * Falls back to Environment resolver if no configuration is found.
     * */
    return ProxyConfigurationResolverProvider::get(request_protocol, configuration);
--- a/src/Common/ProxyConfigurationResolverProvider.h
+++ b/src/Common/ProxyConfigurationResolverProvider.h
@ -33,12 +33,11 @@ public:
    );

 private:
-    template <bool is_new_syntax = true>
    static std::shared_ptr<ProxyConfigurationResolver> getFromSettings(
+        bool is_new_syntax,
        Protocol protocol,
        const String & config_prefix,
-        const Poco::Util::AbstractConfiguration & configuration
-    );
+        const Poco::Util::AbstractConfiguration & configuration);
 };

 }
--- a/src/Common/SignalHandlers.cpp
+++ b/src/Common/SignalHandlers.cpp
@ -18,13 +18,17 @@

 namespace DB
 {
+
 namespace ErrorCodes
 {
 extern const int CANNOT_SET_SIGNAL_HANDLER;
 extern const int CANNOT_SEND_SIGNAL;
 }
+
 }

+extern const char * GIT_HASH;
+
 using namespace DB;


@ -334,7 +338,7 @@ void SignalListener::onTerminate(std::string_view message, UInt32 thread_num) co
    size_t pos = message.find('\n');

    LOG_FATAL(log, "(version {}{}, build id: {}, git hash: {}) (from thread {}) {}",
-              VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", daemon ? daemon->git_hash : "", thread_num, message.substr(0, pos));
+              VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", GIT_HASH, thread_num, message.substr(0, pos));

    /// Print trace from std::terminate exception line-by-line to make it easy for grep.
    while (pos != std::string_view::npos)
@ -368,7 +372,7 @@ try

    LOG_FATAL(log, "########## Short fault info ############");
    LOG_FATAL(log, "(version {}{}, build id: {}, git hash: {}) (from thread {}) Received signal {}",
-              VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", daemon ? daemon->git_hash : "",
+              VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", GIT_HASH,
              thread_num, sig);

    std::string signal_description = "Unknown signal";
@ -434,13 +438,13 @@ try
    if (query_id.empty())
    {
        LOG_FATAL(log, "(version {}{}, build id: {}, git hash: {}) (from thread {}) (no query) Received signal {} ({})",
-                  VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", daemon ? daemon->git_hash : "",
+                  VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", GIT_HASH,
                  thread_num, signal_description, sig);
    }
    else
    {
        LOG_FATAL(log, "(version {}{}, build id: {}, git hash: {}) (from thread {}) (query_id: {}) (query: {}) Received signal {} ({})",
-                  VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", daemon ? daemon->git_hash : "",
+                  VERSION_STRING, VERSION_OFFICIAL, daemon ? daemon->build_id : "", GIT_HASH,
                  thread_num, query_id, query, signal_description, sig);
    }

--- a/src/Core/ExternalTable.cpp
+++ b/src/Core/ExternalTable.cpp
@ -85,7 +85,7 @@ void BaseExternalTable::parseStructureFromStructureField(const std::string & arg
        /// We use `formatWithPossiblyHidingSensitiveData` instead of `getColumnNameWithoutAlias` because `column->type` is an ASTFunction.
        /// `getColumnNameWithoutAlias` will return name of the function with `(arguments)` even if arguments is empty.
        if (column)
-            structure.emplace_back(column->name, column->type->formatWithPossiblyHidingSensitiveData(0, true, true));
+            structure.emplace_back(column->name, column->type->formatWithPossiblyHidingSensitiveData(0, true, true, false));
        else
            throw Exception(ErrorCodes::BAD_ARGUMENTS, "Error while parsing table structure: expected column definition, got {}", child->formatForErrorMessage());
    }
@ -102,7 +102,7 @@ void BaseExternalTable::parseStructureFromTypesField(const std::string & argumen
        throw Exception(ErrorCodes::BAD_ARGUMENTS, "Error while parsing table structure: {}", error);

    for (size_t i = 0; i < type_list_raw->children.size(); ++i)
-        structure.emplace_back("_" + toString(i + 1), type_list_raw->children[i]->formatWithPossiblyHidingSensitiveData(0, true, true));
+        structure.emplace_back("_" + toString(i + 1), type_list_raw->children[i]->formatWithPossiblyHidingSensitiveData(0, true, true, false));
 }

 void BaseExternalTable::initSampleBlock()
--- a/src/Core/Settings.h
+++ b/src/Core/Settings.h
@ -882,7 +882,7 @@ class IColumn;
    M(Bool, use_json_alias_for_old_object_type, false, "When enabled, JSON type alias will create old experimental Object type instead of a new JSON type", 0) \
    M(Bool, allow_create_index_without_type, false, "Allow CREATE INDEX query without TYPE. Query will be ignored. Made for SQL compatibility tests.", 0) \
    M(Bool, create_index_ignore_unique, false, "Ignore UNIQUE keyword in CREATE UNIQUE INDEX. Made for SQL compatibility tests.", 0) \
-    M(Bool, print_pretty_type_names, true, "Print pretty type names in DESCRIBE query and toTypeName() function", 0) \
+    M(Bool, print_pretty_type_names, true, "Print pretty type names in the DESCRIBE query and `toTypeName` function, as well as in the `SHOW CREATE TABLE` query and the `formatQuery` function.", 0) \
    M(Bool, create_table_empty_primary_key_by_default, false, "Allow to create *MergeTree tables with empty primary key when ORDER BY and PRIMARY KEY not specified", 0) \
    M(Bool, allow_named_collection_override_by_default, true, "Allow named collections' fields override by default.", 0) \
    M(SQLSecurityType, default_normal_view_sql_security, SQLSecurityType::INVOKER, "Allows to set a default value for SQL SECURITY option when creating a normal view.", 0) \
--- a/src/Daemon/BaseDaemon.cpp
+++ b/src/Daemon/BaseDaemon.cpp
@ -452,8 +452,6 @@ void BaseDaemon::initializeTerminationAndSignalProcessing()
    build_id = SymbolIndex::instance().getBuildIDHex();
 #endif

-    git_hash = GIT_HASH;
-
 #if defined(OS_LINUX)
    std::string executable_path = getExecutablePath();

@ -466,7 +464,7 @@ void BaseDaemon::logRevision() const
 {
    logger().information("Starting " + std::string{VERSION_FULL}
        + " (revision: " + std::to_string(ClickHouseRevision::getVersionRevision())
-        + ", git hash: " + (git_hash.empty() ? "<unknown>" : git_hash)
+        + ", git hash: " + std::string(GIT_HASH)
        + ", build id: " + (build_id.empty() ? "<unknown>" : build_id) + ")"
        + ", PID " + std::to_string(getpid()));
 }
--- a/src/Daemon/BaseDaemon.h
+++ b/src/Daemon/BaseDaemon.h
@ -165,7 +165,6 @@ protected:
    Poco::Util::AbstractConfiguration * last_configuration = nullptr;

    String build_id;
-    String git_hash;
    String stored_binary_hash;

    bool should_setup_watchdog = false;
--- a/src/DataTypes/IDataType.cpp
+++ b/src/DataTypes/IDataType.cpp
@ -8,7 +8,6 @@
 #include <Common/quoteString.h>

 #include <IO/WriteHelpers.h>
-#include <IO/Operators.h>

 #include <DataTypes/IDataType.h>
 #include <DataTypes/DataTypeCustom.h>
--- a/src/DataTypes/Serializations/SerializationDynamic.cpp
+++ b/src/DataTypes/Serializations/SerializationDynamic.cpp
@ -115,7 +115,7 @@ void SerializationDynamic::serializeBinaryBulkStatePrefix(
    dynamic_state->max_dynamic_types = column_dynamic.getMaxDynamicTypes();
    /// Write max_dynamic_types parameter, because it can differ from the max_dynamic_types
    /// that is specified in the Dynamic type (we could decrease it before merge).
-    writeBinaryLittleEndian(dynamic_state->max_dynamic_types, *stream);
+    writeVarUInt(dynamic_state->max_dynamic_types, *stream);

    dynamic_state->variant_type = variant_info.variant_type;
    dynamic_state->variant_names = variant_info.variant_names;
@ -123,7 +123,7 @@ void SerializationDynamic::serializeBinaryBulkStatePrefix(

    /// Write information about variants.
    size_t num_variants = dynamic_state->variant_names.size() - 1; /// Don't write shared variant, Dynamic column should always have it.
-    writeBinaryLittleEndian(num_variants, *stream);
+    writeVarUInt(num_variants, *stream);
    if (settings.data_types_binary_encoding)
    {
        const auto & variants = assert_cast<const DataTypeVariant &>(*dynamic_state->variant_type).getVariants();
@ -252,11 +252,11 @@ ISerialization::DeserializeBinaryBulkStatePtr SerializationDynamic::deserializeD
        readBinaryLittleEndian(structure_version, *structure_stream);
        auto structure_state = std::make_shared<DeserializeBinaryBulkStateDynamicStructure>(structure_version);
        /// Read max_dynamic_types parameter.
-        readBinaryLittleEndian(structure_state->max_dynamic_types, *structure_stream);
+        readVarUInt(structure_state->max_dynamic_types, *structure_stream);
        /// Read information about variants.
        DataTypes variants;
        size_t num_variants;
-        readBinaryLittleEndian(num_variants, *structure_stream);
+        readVarUInt(num_variants, *structure_stream);
        variants.reserve(num_variants + 1); /// +1 for shared variant.
        if (settings.data_types_binary_encoding)
        {
--- a/src/DataTypes/Serializations/SerializationObject.cpp
+++ b/src/DataTypes/Serializations/SerializationObject.cpp
@ -199,7 +199,7 @@ void SerializationObject::serializeBinaryBulkStatePrefix(
    auto object_state = std::make_shared<SerializeBinaryBulkStateObject>(serialization_version);
    object_state->max_dynamic_paths = column_object.getMaxDynamicPaths();
    /// Write max_dynamic_paths parameter.
-    writeBinaryLittleEndian(object_state->max_dynamic_paths, *stream);
+    writeVarUInt(object_state->max_dynamic_paths, *stream);
    /// Write all dynamic paths in sorted order.
    object_state->sorted_dynamic_paths.reserve(dynamic_paths.size());
    for (const auto & [path, _] : dynamic_paths)
@ -354,7 +354,7 @@ ISerialization::DeserializeBinaryBulkStatePtr SerializationObject::deserializeOb
        readBinaryLittleEndian(serialization_version, *structure_stream);
        auto structure_state = std::make_shared<DeserializeBinaryBulkStateObjectStructure>(serialization_version);
        /// Read max_dynamic_paths parameter.
-        readBinaryLittleEndian(structure_state->max_dynamic_paths, *structure_stream);
+        readVarUInt(structure_state->max_dynamic_paths, *structure_stream);
        /// Read the sorted list of dynamic paths.
        size_t dynamic_paths_size;
        readVarUInt(dynamic_paths_size, *structure_stream);
--- a/src/Databases/DatabaseOnDisk.cpp
+++ b/src/Databases/DatabaseOnDisk.cpp
@ -504,7 +504,7 @@ void DatabaseOnDisk::renameTable(
 }


-/// It returns create table statement (even if table is detached)
+/// It returns the create table statement (even if table is detached)
 ASTPtr DatabaseOnDisk::getCreateTableQueryImpl(const String & table_name, ContextPtr, bool throw_on_error) const
 {
    ASTPtr ast;
--- a/src/Functions/formatQuery.cpp
+++ b/src/Functions/formatQuery.cpp
@ -43,6 +43,7 @@ public:
        max_query_size = settings.max_query_size;
        max_parser_depth = settings.max_parser_depth;
        max_parser_backtracks = settings.max_parser_backtracks;
+        print_pretty_type_names = settings.print_pretty_type_names;
    }

    String getName() const override { return name; }
@ -138,7 +139,11 @@ private:
                }
            }

-            formatAST(*ast, buf, /*hilite*/ false, /*single_line*/ output_formatting == OutputFormatting::SingleLine);
+            IAST::FormatSettings settings(buf, output_formatting == OutputFormatting::SingleLine, /*hilite*/ false);
+            settings.show_secrets = true;
+            settings.print_pretty_type_names = print_pretty_type_names;
+            ast->format(settings);
+
            auto formatted = buf.stringView();

            const size_t res_data_new_size = res_data_size + formatted.size() + 1;
@ -165,6 +170,7 @@ private:
    size_t max_query_size;
    size_t max_parser_depth;
    size_t max_parser_backtracks;
+    bool print_pretty_type_names;
 };

 }
--- a/src/IO/ReadWriteBufferFromHTTP.cpp
+++ b/src/IO/ReadWriteBufferFromHTTP.cpp
@ -238,7 +238,7 @@ ReadWriteBufferFromHTTP::ReadWriteBufferFromHTTP(

    if (iter == http_header_entries.end())
    {
-        http_header_entries.emplace_back(user_agent, fmt::format("ClickHouse/{}", VERSION_STRING));
+        http_header_entries.emplace_back(user_agent, fmt::format("ClickHouse/{}{}", VERSION_STRING, VERSION_OFFICIAL));
    }

    if (!delay_initialization && use_external_buffer)
--- a/src/IO/S3/Client.cpp
+++ b/src/IO/S3/Client.cpp
@ -982,10 +982,10 @@ PocoHTTPClientConfiguration ClientFactory::createClientConfiguration( // NOLINT
 {
    auto context = Context::getGlobalContextInstance();
    chassert(context);
-    auto proxy_configuration_resolver = DB::ProxyConfigurationResolverProvider::get(DB::ProxyConfiguration::protocolFromString(protocol), context->getConfigRef());
+    auto proxy_configuration_resolver = ProxyConfigurationResolverProvider::get(ProxyConfiguration::protocolFromString(protocol), context->getConfigRef());

-    auto per_request_configuration = [=] () { return proxy_configuration_resolver->resolve(); };
-    auto error_report = [=] (const DB::ProxyConfiguration & req) { proxy_configuration_resolver->errorReport(req); };
+    auto per_request_configuration = [=]{ return proxy_configuration_resolver->resolve(); };
+    auto error_report = [=](const ProxyConfiguration & req) { proxy_configuration_resolver->errorReport(req); };

    auto config = PocoHTTPClientConfiguration(
        per_request_configuration,
--- a/src/IO/S3/PocoHTTPClient.cpp
+++ b/src/IO/S3/PocoHTTPClient.cpp
@ -1,4 +1,5 @@
 #include <Poco/Timespan.h>
+#include <Common/config_version.h>
 #include "config.h"

 #if USE_AWS_S3
@ -17,6 +18,7 @@
 #include <IO/WriteBufferFromString.h>
 #include <IO/Operators.h>
 #include <IO/S3/ProviderType.h>
+#include <Interpreters/Context.h>

 #include <aws/core/http/HttpRequest.h>
 #include <aws/core/http/HttpResponse.h>
@ -29,6 +31,7 @@

 #include <boost/algorithm/string.hpp>

+
 static const int SUCCESS_RESPONSE_MIN = 200;
 static const int SUCCESS_RESPONSE_MAX = 299;

@ -84,7 +87,7 @@ namespace DB::S3
 {

 PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
-        std::function<DB::ProxyConfiguration()> per_request_configuration_,
+        std::function<ProxyConfiguration()> per_request_configuration_,
        const String & force_region_,
        const RemoteHostFilter & remote_host_filter_,
        unsigned int s3_max_redirects_,
@ -94,7 +97,7 @@ PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
        bool s3_use_adaptive_timeouts_,
        const ThrottlerPtr & get_request_throttler_,
        const ThrottlerPtr & put_request_throttler_,
-        std::function<void(const DB::ProxyConfiguration &)> error_report_)
+        std::function<void(const ProxyConfiguration &)> error_report_)
    : per_request_configuration(per_request_configuration_)
    , force_region(force_region_)
    , remote_host_filter(remote_host_filter_)
@ -107,6 +110,8 @@ PocoHTTPClientConfiguration::PocoHTTPClientConfiguration(
    , s3_use_adaptive_timeouts(s3_use_adaptive_timeouts_)
    , error_report(error_report_)
 {
+    /// This is used to identify configurations created by us.
+    userAgent = std::string(VERSION_FULL) + VERSION_OFFICIAL;
 }

 void PocoHTTPClientConfiguration::updateSchemeAndRegion()
@ -166,6 +171,17 @@ PocoHTTPClient::PocoHTTPClient(const PocoHTTPClientConfiguration & client_config
 {
 }

+PocoHTTPClient::PocoHTTPClient(const Aws::Client::ClientConfiguration & client_configuration)
+    : timeouts(ConnectionTimeouts()
+       .withConnectionTimeout(Poco::Timespan(client_configuration.connectTimeoutMs * 1000))
+       .withSendTimeout(Poco::Timespan(client_configuration.requestTimeoutMs * 1000))
+       .withReceiveTimeout(Poco::Timespan(client_configuration.requestTimeoutMs * 1000))
+       .withTCPKeepAliveTimeout(Poco::Timespan(
+           client_configuration.enableTcpKeepAlive ? client_configuration.tcpKeepAliveIntervalMs * 1000 : 0))),
+    remote_host_filter(Context::getGlobalContextInstance()->getRemoteHostFilter())
+{
+}
+
 std::shared_ptr<Aws::Http::HttpResponse> PocoHTTPClient::MakeRequest(
    const std::shared_ptr<Aws::Http::HttpRequest> & request,
    Aws::Utils::RateLimits::RateLimiterInterface * readLimiter,
@ -381,8 +397,11 @@ void PocoHTTPClient::makeRequestInternalImpl(

    try
    {
-        const auto proxy_configuration = per_request_configuration();
-        for (unsigned int attempt = 0; attempt <= s3_max_redirects; ++attempt)
+        ProxyConfiguration proxy_configuration;
+        if (per_request_configuration)
+            proxy_configuration = per_request_configuration();
+
+        for (size_t attempt = 0; attempt <= s3_max_redirects; ++attempt)
        {
            Poco::URI target_uri(uri);

@ -500,7 +519,6 @@ void PocoHTTPClient::makeRequestInternalImpl(
                    LOG_TEST(log, "Redirecting request to new location: {}", location);

                addMetric(request, S3MetricType::Redirects);
-
                continue;
            }

@ -548,9 +566,9 @@ void PocoHTTPClient::makeRequestInternalImpl(
            }
            else
            {
-
                if (status_code == 429 || status_code == 503)
-                { // API throttling
+                {
+                    /// API throttling
                    addMetric(request, S3MetricType::Throttling);
                }
                else if (status_code >= 300)
--- a/src/IO/S3/PocoHTTPClient.h
+++ b/src/IO/S3/PocoHTTPClient.h
@ -20,6 +20,7 @@
 #include <aws/core/http/HttpRequest.h>
 #include <aws/core/http/standard/StandardHttpResponse.h>

+
 namespace Aws::Http::Standard
 {
 class StandardHttpResponse;
@ -27,18 +28,20 @@ class StandardHttpResponse;

 namespace DB
 {
-
 class Context;
 }

+
 namespace DB::S3
 {
+
 class ClientFactory;
 class PocoHTTPClient;

+
 struct PocoHTTPClientConfiguration : public Aws::Client::ClientConfiguration
 {
-    std::function<DB::ProxyConfiguration()> per_request_configuration;
+    std::function<ProxyConfiguration()> per_request_configuration;
    String force_region;
    const RemoteHostFilter & remote_host_filter;
    unsigned int s3_max_redirects;
@ -54,13 +57,13 @@ struct PocoHTTPClientConfiguration : public Aws::Client::ClientConfiguration
    size_t http_keep_alive_timeout = DEFAULT_HTTP_KEEP_ALIVE_TIMEOUT;
    size_t http_keep_alive_max_requests = DEFAULT_HTTP_KEEP_ALIVE_MAX_REQUEST;

-    std::function<void(const DB::ProxyConfiguration &)> error_report;
+    std::function<void(const ProxyConfiguration &)> error_report;

    void updateSchemeAndRegion();

 private:
    PocoHTTPClientConfiguration(
-        std::function<DB::ProxyConfiguration()> per_request_configuration_,
+        std::function<ProxyConfiguration()> per_request_configuration_,
        const String & force_region_,
        const RemoteHostFilter & remote_host_filter_,
        unsigned int s3_max_redirects_,
@ -70,13 +73,13 @@ private:
        bool s3_use_adaptive_timeouts_,
        const ThrottlerPtr & get_request_throttler_,
        const ThrottlerPtr & put_request_throttler_,
-        std::function<void(const DB::ProxyConfiguration &)> error_report_
-    );
+        std::function<void(const ProxyConfiguration &)> error_report_);

    /// Constructor of Aws::Client::ClientConfiguration must be called after AWS SDK initialization.
    friend ClientFactory;
 };

+
 class PocoHTTPResponse : public Aws::Http::Standard::StandardHttpResponse
 {
 public:
@ -116,10 +119,12 @@ private:
    Aws::Utils::Stream::ResponseStream body_stream;
 };

+
 class PocoHTTPClient : public Aws::Http::HttpClient
 {
 public:
    explicit PocoHTTPClient(const PocoHTTPClientConfiguration & client_configuration);
+    explicit PocoHTTPClient(const Aws::Client::ClientConfiguration & client_configuration);
    ~PocoHTTPClient() override = default;

    std::shared_ptr<Aws::Http::HttpResponse> MakeRequest(
@ -166,14 +171,14 @@ protected:
    static S3MetricKind getMetricKind(const Aws::Http::HttpRequest & request);
    void addMetric(const Aws::Http::HttpRequest & request, S3MetricType type, ProfileEvents::Count amount = 1) const;

-    std::function<DB::ProxyConfiguration()> per_request_configuration;
-    std::function<void(const DB::ProxyConfiguration &)> error_report;
+    std::function<ProxyConfiguration()> per_request_configuration;
+    std::function<void(const ProxyConfiguration &)> error_report;
    ConnectionTimeouts timeouts;
    const RemoteHostFilter & remote_host_filter;
-    unsigned int s3_max_redirects;
+    unsigned int s3_max_redirects = 0;
    bool s3_use_adaptive_timeouts = true;
-    bool enable_s3_requests_logging;
-    bool for_disk_s3;
+    bool enable_s3_requests_logging = false;
+    bool for_disk_s3 = false;

    /// Limits get request per second rate for GET, SELECT and all other requests, excluding throttled by put throttler
    /// (i.e. throttles GetObject, HeadObject)
--- a/src/IO/S3/PocoHTTPClientFactory.cpp
+++ b/src/IO/S3/PocoHTTPClientFactory.cpp
@ -15,7 +15,10 @@ namespace DB::S3
 std::shared_ptr<Aws::Http::HttpClient>
 PocoHTTPClientFactory::CreateHttpClient(const Aws::Client::ClientConfiguration & client_configuration) const
 {
-    return std::make_shared<PocoHTTPClient>(static_cast<const PocoHTTPClientConfiguration &>(client_configuration));
+    if (client_configuration.userAgent.starts_with("ClickHouse"))
+        return std::make_shared<PocoHTTPClient>(static_cast<const PocoHTTPClientConfiguration &>(client_configuration));
+    else /// This client is created inside the AWS SDK with default settings to obtain ECS credentials from localhost.
+        return std::make_shared<PocoHTTPClient>(client_configuration);
 }

 std::shared_ptr<Aws::Http::HttpRequest> PocoHTTPClientFactory::CreateHttpRequest(
--- a/src/Interpreters/InterpreterSetQuery.cpp
+++ b/src/Interpreters/InterpreterSetQuery.cpp
@ -9,6 +9,7 @@
 #include <Parsers/ASTQueryWithOutput.h>
 #include <Parsers/ASTSelectWithUnionQuery.h>

+
 namespace DB
 {

@ -45,9 +46,7 @@ static void applySettingsFromSelectWithUnion(const ASTSelectWithUnionQuery & sel
    // It is flattened later, when we process UNION ALL/DISTINCT.
    const auto * last_select = children.back()->as<ASTSelectQuery>();
    if (last_select && last_select->settings())
-    {
-        InterpreterSetQuery(last_select->settings(), context).executeForCurrentContext();
-    }
+        InterpreterSetQuery(last_select->settings(), context).executeForCurrentContext(/* ignore_setting_constraints= */ false);
 }

 void InterpreterSetQuery::applySettingsFromQuery(const ASTPtr & ast, ContextMutablePtr context_)
@ -55,10 +54,20 @@ void InterpreterSetQuery::applySettingsFromQuery(const ASTPtr & ast, ContextMuta
    if (!ast)
        return;

+    /// First apply the outermost settings. Then they could be overridden by deeper settings.
+    if (const auto * query_with_output = dynamic_cast<const ASTQueryWithOutput *>(ast.get()))
+    {
+        if (query_with_output->settings_ast)
+            InterpreterSetQuery(query_with_output->settings_ast, context_).executeForCurrentContext(/* ignore_setting_constraints= */ false);
+
+        if (const auto * create_query = ast->as<ASTCreateQuery>(); create_query && create_query->select)
+            applySettingsFromSelectWithUnion(create_query->select->as<ASTSelectWithUnionQuery &>(), context_);
+    }
+
    if (const auto * select_query = ast->as<ASTSelectQuery>())
    {
        if (auto new_settings = select_query->settings())
-            InterpreterSetQuery(new_settings, context_).executeForCurrentContext();
+            InterpreterSetQuery(new_settings, context_).executeForCurrentContext(/* ignore_setting_constraints= */ false);
    }
    else if (const auto * select_with_union_query = ast->as<ASTSelectWithUnionQuery>())
    {
@ -67,28 +76,15 @@ void InterpreterSetQuery::applySettingsFromQuery(const ASTPtr & ast, ContextMuta
    else if (const auto * explain_query = ast->as<ASTExplainQuery>())
    {
        if (explain_query->settings_ast)
-            InterpreterSetQuery(explain_query->settings_ast, context_).executeForCurrentContext();
+            InterpreterSetQuery(explain_query->settings_ast, context_).executeForCurrentContext(/* ignore_setting_constraints= */ false);

        applySettingsFromQuery(explain_query->getExplainedQuery(), context_);
    }
-    else if (const auto * query_with_output = dynamic_cast<const ASTQueryWithOutput *>(ast.get()))
-    {
-        if (query_with_output->settings_ast)
-            InterpreterSetQuery(query_with_output->settings_ast, context_).executeForCurrentContext();
-
-        if (const auto * create_query = ast->as<ASTCreateQuery>())
-        {
-            if (create_query->select)
-            {
-                applySettingsFromSelectWithUnion(create_query->select->as<ASTSelectWithUnionQuery &>(), context_);
-            }
-        }
-    }
    else if (auto * insert_query = ast->as<ASTInsertQuery>())
    {
        context_->setInsertFormat(insert_query->format);
        if (insert_query->settings_ast)
-            InterpreterSetQuery(insert_query->settings_ast, context_).executeForCurrentContext();
+            InterpreterSetQuery(insert_query->settings_ast, context_).executeForCurrentContext(/* ignore_setting_constraints= */ false);
    }
 }

--- a/src/Interpreters/InterpreterSetQuery.h
+++ b/src/Interpreters/InterpreterSetQuery.h
@ -23,7 +23,7 @@ public:
    /** Set setting for current context (query context).
      * It is used for interpretation of SETTINGS clause in SELECT query.
      */
-    void executeForCurrentContext(bool ignore_setting_constraints = false);
+    void executeForCurrentContext(bool ignore_setting_constraints);

    bool supportsTransactions() const override { return true; }

--- a/src/Interpreters/InterpreterShowCreateQuery.cpp
+++ b/src/Interpreters/InterpreterShowCreateQuery.cpp
@ -97,7 +97,12 @@ QueryPipeline InterpreterShowCreateQuery::executeImpl()
    }

    MutableColumnPtr column = ColumnString::create();
-    column->insert(format({.ctx = getContext(), .query = *create_query, .one_line = false}));
+    column->insert(format(
+    {
+        .ctx = getContext(),
+        .query = *create_query,
+        .one_line = false
+    }));

    return QueryPipeline(std::make_shared<SourceFromSingleChunk>(Block{{
        std::move(column),
--- a/src/Interpreters/executeQuery.cpp
+++ b/src/Interpreters/executeQuery.cpp
@ -786,7 +786,7 @@ static std::tuple<ASTPtr, BlockIO> executeQueryImpl(
            /// Verify that AST formatting is consistent:
            /// If you format AST, parse it back, and format it again, you get the same string.

-            String formatted1 = ast->formatWithPossiblyHidingSensitiveData(0, true, true);
+            String formatted1 = ast->formatWithPossiblyHidingSensitiveData(0, true, true, false);

            /// The query can become more verbose after formatting, so:
            size_t new_max_query_size = max_query_size > 0 ? (1000 + 2 * max_query_size) : 0;
@ -811,7 +811,7 @@ static std::tuple<ASTPtr, BlockIO> executeQueryImpl(

            chassert(ast2);

-            String formatted2 = ast2->formatWithPossiblyHidingSensitiveData(0, true, true);
+            String formatted2 = ast2->formatWithPossiblyHidingSensitiveData(0, true, true, false);

            if (formatted1 != formatted2)
                throw Exception(ErrorCodes::LOGICAL_ERROR,
--- a/src/Interpreters/formatWithPossiblyHidingSecrets.h
+++ b/src/Interpreters/formatWithPossiblyHidingSecrets.h
@ -25,7 +25,8 @@ inline String format(const SecretHidingFormatSettings & settings)
        && settings.ctx->getSettingsRef().format_display_secrets_in_show_and_select
        && settings.ctx->getAccess()->isGranted(AccessType::displaySecretsInShowAndSelect);

-    return settings.query.formatWithPossiblyHidingSensitiveData(settings.max_length, settings.one_line, show_secrets);
+    return settings.query.formatWithPossiblyHidingSensitiveData(
+        settings.max_length, settings.one_line, show_secrets, settings.ctx->getSettingsRef().print_pretty_type_names);
 }

 }
--- a/src/Parsers/ASTColumnDeclaration.cpp
+++ b/src/Parsers/ASTColumnDeclaration.cpp
@ -66,17 +66,13 @@ void ASTColumnDeclaration::formatImpl(const FormatSettings & format_settings, Fo
 {
    frame.need_parens = false;

-    /// We have to always backquote column names to avoid ambiguouty with INDEX and other declarations in CREATE query.
+    /// We have to always backquote column names to avoid ambiguity with INDEX and other declarations in CREATE query.
    format_settings.ostr << backQuote(name);

    if (type)
    {
        format_settings.ostr << ' ';
-
-        FormatStateStacked type_frame = frame;
-        type_frame.indent = 0;
-
-        type->formatImpl(format_settings, state, type_frame);
+        type->formatImpl(format_settings, state, frame);
    }

    if (null_modifier)
--- a/src/Parsers/ASTDataType.cpp
+++ b/src/Parsers/ASTDataType.cpp
@ -40,12 +40,22 @@ void ASTDataType::formatImpl(const FormatSettings & settings, FormatState & stat
    {
        settings.ostr << '(' << (settings.hilite ? hilite_none : "");

-        for (size_t i = 0, size = arguments->children.size(); i < size; ++i)
+        if (!settings.one_line && settings.print_pretty_type_names && name == "Tuple")
        {
-            if (i != 0)
-                settings.ostr << ", ";
-
-            arguments->children[i]->formatImpl(settings, state, frame);
+            ++frame.indent;
+            std::string indent_str = settings.one_line ? "" : "\n" + std::string(4 * frame.indent, ' ');
+            for (size_t i = 0, size = arguments->children.size(); i < size; ++i)
+            {
+                if (i != 0)
+                    settings.ostr << ',';
+                settings.ostr << indent_str;
+                arguments->children[i]->formatImpl(settings, state, frame);
+            }
+        }
+        else
+        {
+            frame.expression_list_prepend_whitespace = false;
+            arguments->formatImpl(settings, state, frame);
        }

        settings.ostr << (settings.hilite ? hilite_function : "") << ')';
--- a/src/Parsers/ASTExpressionList.cpp
+++ b/src/Parsers/ASTExpressionList.cpp
@ -42,7 +42,8 @@ void ASTExpressionList::formatImpl(const FormatSettings & settings, FormatState

 void ASTExpressionList::formatImplMultiline(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
 {
-    std::string indent_str = "\n" + std::string(4 * (frame.indent + 1), ' ');
+    ++frame.indent;
+    std::string indent_str = "\n" + std::string(4 * frame.indent, ' ');

    if (frame.expression_list_prepend_whitespace)
    {
@ -50,8 +51,6 @@ void ASTExpressionList::formatImplMultiline(const FormatSettings & settings, For
            settings.ostr << ' ';
    }

-    ++frame.indent;
-
    for (size_t i = 0, size = children.size(); i < size; ++i)
    {
        if (i && separator)
--- a/src/Parsers/ASTNameTypePair.cpp
+++ b/src/Parsers/ASTNameTypePair.cpp
@ -23,12 +23,8 @@ ASTPtr ASTNameTypePair::clone() const

 void ASTNameTypePair::formatImpl(const FormatSettings & settings, FormatState & state, FormatStateStacked frame) const
 {
-    std::string indent_str = settings.one_line ? "" : std::string(4 * frame.indent, ' ');
-
-    settings.ostr << indent_str << backQuoteIfNeed(name) << ' ';
+    settings.ostr << backQuoteIfNeed(name) << ' ';
    type->formatImpl(settings, state, frame);
 }

 }
-
-
--- a/src/Parsers/ExpressionElementParsers.cpp
+++ b/src/Parsers/ExpressionElementParsers.cpp
@ -853,9 +853,10 @@ bool ParserCastOperator::parseImpl(Pos & pos, ASTPtr & node, Expected & expected

    /// Parse numbers (including decimals), strings, arrays and tuples of them.

+    Pos begin = pos;
    const char * data_begin = pos->begin;
    const char * data_end = pos->end;
-    bool is_string_literal = pos->type == StringLiteral;
+    ASTPtr string_literal;

    if (pos->type == Minus)
    {
@ -866,10 +867,15 @@ bool ParserCastOperator::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
        data_end = pos->end;
        ++pos;
    }
-    else if (pos->type == Number || is_string_literal)
+    else if (pos->type == Number)
    {
        ++pos;
    }
+    else if (pos->type == StringLiteral)
+    {
+        if (!ParserStringLiteral().parse(begin, string_literal, expected))
+            return false;
+    }
    else if (isOneOf<OpeningSquareBracket, OpeningRoundBracket>(pos->type))
    {
        TokenType last_token = OpeningSquareBracket;
@ -937,20 +943,18 @@ bool ParserCastOperator::parseImpl(Pos & pos, ASTPtr & node, Expected & expected
    if (ParserToken(DoubleColon).ignore(pos, expected)
        && ParserDataType().parse(pos, type_ast, expected))
    {
-        String s;
        size_t data_size = data_end - data_begin;
-        if (is_string_literal)
+        if (string_literal)
        {
-            ReadBufferFromMemory buf(data_begin, data_size);
-            readQuotedStringWithSQLStyle(s, buf);
-            assert(buf.count() == data_size);
+            node = createFunctionCast(string_literal, type_ast);
+            return true;
        }
        else
-            s = String(data_begin, data_size);
-
-        auto literal = std::make_shared<ASTLiteral>(std::move(s));
-        node = createFunctionCast(literal, type_ast);
-        return true;
+        {
+            auto literal = std::make_shared<ASTLiteral>(String(data_begin, data_size));
+            node = createFunctionCast(literal, type_ast);
+            return true;
+        }
    }

    return false;
--- a/src/Parsers/IAST.cpp
+++ b/src/Parsers/IAST.cpp
@ -165,11 +165,12 @@ size_t IAST::checkDepthImpl(size_t max_depth) const
    return res;
 }

-String IAST::formatWithPossiblyHidingSensitiveData(size_t max_length, bool one_line, bool show_secrets) const
+String IAST::formatWithPossiblyHidingSensitiveData(size_t max_length, bool one_line, bool show_secrets, bool print_pretty_type_names) const
 {
    WriteBufferFromOwnString buf;
    FormatSettings settings(buf, one_line);
    settings.show_secrets = show_secrets;
+    settings.print_pretty_type_names = print_pretty_type_names;
    format(settings);
    return wipeSensitiveDataAndCutToLength(buf.str(), max_length);
 }
--- a/src/Parsers/IAST.h
+++ b/src/Parsers/IAST.h
@ -201,6 +201,7 @@ public:
        bool show_secrets; /// Show secret parts of the AST (e.g. passwords, encryption keys).
        char nl_or_ws; /// Newline or whitespace.
        LiteralEscapingStyle literal_escaping_style;
+        bool print_pretty_type_names;

        explicit FormatSettings(
            WriteBuffer & ostr_,
@ -209,7 +210,8 @@ public:
            bool always_quote_identifiers_ = false,
            IdentifierQuotingStyle identifier_quoting_style_ = IdentifierQuotingStyle::Backticks,
            bool show_secrets_ = true,
-            LiteralEscapingStyle literal_escaping_style_ = LiteralEscapingStyle::Regular)
+            LiteralEscapingStyle literal_escaping_style_ = LiteralEscapingStyle::Regular,
+            bool print_pretty_type_names_ = false)
            : ostr(ostr_)
            , one_line(one_line_)
            , hilite(hilite_)
@ -218,6 +220,7 @@ public:
            , show_secrets(show_secrets_)
            , nl_or_ws(one_line ? ' ' : '\n')
            , literal_escaping_style(literal_escaping_style_)
+            , print_pretty_type_names(print_pretty_type_names_)
        {
        }

@ -230,6 +233,7 @@ public:
            , show_secrets(other.show_secrets)
            , nl_or_ws(other.nl_or_ws)
            , literal_escaping_style(other.literal_escaping_style)
+            , print_pretty_type_names(other.print_pretty_type_names)
        {
        }

@ -251,7 +255,7 @@ public:
    /// The state that is copied when each node is formatted. For example, nesting level.
    struct FormatStateStacked
    {
-        UInt8 indent = 0;
+        UInt16 indent = 0;
        bool need_parens = false;
        bool expression_list_always_start_on_new_line = false;  /// Line feed and indent before expression list even if it's of single element.
        bool expression_list_prepend_whitespace = false; /// Prepend whitespace (if it is required)
@ -274,7 +278,7 @@ public:

    /// Secrets are displayed regarding show_secrets, then SensitiveDataMasker is applied.
    /// You can use Interpreters/formatWithPossiblyHidingSecrets.h for convenience.
-    String formatWithPossiblyHidingSensitiveData(size_t max_length, bool one_line, bool show_secrets) const;
+    String formatWithPossiblyHidingSensitiveData(size_t max_length, bool one_line, bool show_secrets, bool print_pretty_type_names) const;

    /** formatForLogging and formatForErrorMessage always hide secrets. This inconsistent
      * behaviour is due to the fact such functions are called from Client which knows nothing about
@ -283,12 +287,12 @@ public:
      */
    String formatForLogging(size_t max_length = 0) const
    {
-        return formatWithPossiblyHidingSensitiveData(max_length, true, false);
+        return formatWithPossiblyHidingSensitiveData(max_length, true, false, false);
    }

    String formatForErrorMessage() const
    {
-        return formatWithPossiblyHidingSensitiveData(0, true, false);
+        return formatWithPossiblyHidingSensitiveData(0, true, false, false);
    }

    virtual bool hasSecretParts() const { return childrenHaveSecretParts(); }
--- a/src/Parsers/ParserQueryWithOutput.cpp
+++ b/src/Parsers/ParserQueryWithOutput.cpp
@ -25,7 +25,6 @@
 #include <Parsers/ParserTablePropertiesQuery.h>
 #include <Parsers/ParserWatchQuery.h>
 #include <Parsers/ParserDescribeCacheQuery.h>
-#include <Parsers/QueryWithOutputSettingsPushDownVisitor.h>
 #include <Parsers/Access/ParserShowAccessEntitiesQuery.h>
 #include <Parsers/Access/ParserShowAccessQuery.h>
 #include <Parsers/Access/ParserShowCreateAccessEntityQuery.h>
@ -152,37 +151,55 @@ bool ParserQueryWithOutput::parseImpl(Pos & pos, ASTPtr & node, Expected & expec

    }

+    /// These two sections are allowed in an arbitrary order.
    ParserKeyword s_format(Keyword::FORMAT);
-
-    if (s_format.ignore(pos, expected))
-    {
-        ParserIdentifier format_p;
-
-        if (!format_p.parse(pos, query_with_output.format, expected))
-            return false;
-        setIdentifierSpecial(query_with_output.format);
-
-        query_with_output.children.push_back(query_with_output.format);
-    }
-
-    // SETTINGS key1 = value1, key2 = value2, ...
    ParserKeyword s_settings(Keyword::SETTINGS);
-    if (!query_with_output.settings_ast && s_settings.ignore(pos, expected))
-    {
-        ParserSetQuery parser_settings(true);
-        if (!parser_settings.parse(pos, query_with_output.settings_ast, expected))
-            return false;
-        query_with_output.children.push_back(query_with_output.settings_ast);

-        // SETTINGS after FORMAT is not parsed by the SELECT parser (ParserSelectQuery)
-        // Pass them manually, to apply in InterpreterSelectQuery::initSettings()
-        if (query->as<ASTSelectWithUnionQuery>())
+    /** Why: let's take the following example:
+      * SELECT 1 UNION ALL SELECT 2 FORMAT TSV
+      * Each subquery can be put in parentheses and have its own settings:
+      *   (SELECT 1 SETTINGS a=b) UNION ALL (SELECT 2 SETTINGS c=d) FORMAT TSV
+      * And the whole query can have settings:
+      *   (SELECT 1 SETTINGS a=b) UNION ALL (SELECT 2 SETTINGS c=d) FORMAT TSV SETTINGS e=f
+      * A single query with output is parsed in the same way as the UNION ALL chain:
+      *   SELECT 1 SETTINGS a=b FORMAT TSV SETTINGS e=f
+      * So while these forms have a slightly different meaning, they both exist:
+      *   SELECT 1 SETTINGS a=b FORMAT TSV
+      *   SELECT 1 FORMAT TSV SETTINGS e=f
+      * And due to this effect, the users expect that the FORMAT and SETTINGS may go in an arbitrary order.
+      * But while this work:
+      *   (SELECT 1) UNION ALL (SELECT 2) FORMAT TSV SETTINGS d=f
+      * This does not work automatically, unless we explicitly allow different orders:
+      *   (SELECT 1) UNION ALL (SELECT 2) SETTINGS d=f FORMAT TSV
+      * Inevitably, we also allow this:
+      *   SELECT 1 SETTINGS a=b SETTINGS d=f FORMAT TSV
+      *   ^^^^^^^^^^^^^^^^^^^^^
+      * Because this part is consumed into ASTSelectWithUnionQuery
+      * and the rest into ASTQueryWithOutput.
+      */
+
+    for (size_t i = 0; i < 2; ++i)
+    {
+        if (!query_with_output.format && s_format.ignore(pos, expected))
        {
-            auto settings = query_with_output.settings_ast->clone();
-            assert_cast<ASTSetQuery *>(settings.get())->print_in_format = false;
-            QueryWithOutputSettingsPushDownVisitor::Data data{settings};
-            QueryWithOutputSettingsPushDownVisitor(data).visit(query);
+            ParserIdentifier format_p;
+
+            if (!format_p.parse(pos, query_with_output.format, expected))
+                return false;
+            setIdentifierSpecial(query_with_output.format);
+
+            query_with_output.children.push_back(query_with_output.format);
        }
+        else if (!query_with_output.settings_ast && s_settings.ignore(pos, expected))
+        {
+            // SETTINGS key1 = value1, key2 = value2, ...
+            ParserSetQuery parser_settings(true);
+            if (!parser_settings.parse(pos, query_with_output.settings_ast, expected))
+                return false;
+            query_with_output.children.push_back(query_with_output.settings_ast);
+        }
+        else
+            break;
    }

    node = std::move(query);
--- a/src/Parsers/QueryWithOutputSettingsPushDownVisitor.cpp
+++ b/src/Parsers/QueryWithOutputSettingsPushDownVisitor.cpp
@ -1,56 +0,0 @@
-#include <Common/SettingsChanges.h>
-#include <Parsers/QueryWithOutputSettingsPushDownVisitor.h>
-#include <Parsers/ASTSelectWithUnionQuery.h>
-#include <Parsers/ASTSelectQuery.h>
-#include <Parsers/ASTSetQuery.h>
-#include <Parsers/ASTSubquery.h>
-
-#include <iterator>
-#include <algorithm>
-
-namespace DB
-{
-
-bool QueryWithOutputSettingsPushDownMatcher::needChildVisit(ASTPtr & node, const ASTPtr & child)
-{
-    if (node->as<ASTSelectWithUnionQuery>())
-        return true;
-    if (node->as<ASTSubquery>())
-        return true;
-    if (child->as<ASTSelectQuery>())
-        return true;
-    return false;
-}
-
-void QueryWithOutputSettingsPushDownMatcher::visit(ASTPtr & ast, Data & data)
-{
-    if (auto * select_query = ast->as<ASTSelectQuery>())
-        visit(*select_query, ast, data);
-}
-
-void QueryWithOutputSettingsPushDownMatcher::visit(ASTSelectQuery & select_query, ASTPtr &, Data & data)
-{
-    ASTPtr select_settings_ast = select_query.settings();
-    if (!select_settings_ast)
-    {
-        select_query.setExpression(ASTSelectQuery::Expression::SETTINGS, data.settings_ast->clone());
-        return;
-    }
-
-    SettingsChanges & select_settings = select_settings_ast->as<ASTSetQuery &>().changes;
-    SettingsChanges & settings = data.settings_ast->as<ASTSetQuery &>().changes;
-
-    for (auto & setting : settings)
-    {
-        auto it = std::find_if(select_settings.begin(), select_settings.end(), [&](auto & select_setting)
-        {
-            return select_setting.name == setting.name;
-        });
-        if (it == select_settings.end())
-            select_settings.push_back(setting);
-        else
-            it->value = setting.value;
-    }
-}
-
-}
--- a/src/Parsers/QueryWithOutputSettingsPushDownVisitor.h
+++ b/src/Parsers/QueryWithOutputSettingsPushDownVisitor.h
@ -1,39 +0,0 @@
-#pragma once
-
-#include <Parsers/IAST.h>
-#include <Interpreters/InDepthNodeVisitor.h>
-
-namespace DB
-{
-
-class ASTSelectQuery;
-struct SettingChange;
-class SettingsChanges;
-
-/// Pushdown SETTINGS clause that goes after FORMAT to the SELECT query:
-/// (since settings after FORMAT parsed separately not in the ParserSelectQuery but in ParserQueryWithOutput)
-///
-///     SELECT 1                             FORMAT Null SETTINGS max_block_size = 1 ->
-///     SELECT 1 SETTINGS max_block_size = 1 FORMAT Null SETTINGS max_block_size = 1
-///
-/// Otherwise settings after FORMAT will not be applied.
-class QueryWithOutputSettingsPushDownMatcher
-{
-public:
-    using Visitor = InDepthNodeVisitor<QueryWithOutputSettingsPushDownMatcher, true>;
-
-    struct Data
-    {
-        const ASTPtr & settings_ast;
-    };
-
-    static bool needChildVisit(ASTPtr & node, const ASTPtr & child);
-    static void visit(ASTPtr & ast, Data & data);
-
-private:
-    static void visit(ASTSelectQuery &, ASTPtr &, Data &);
-};
-
-using QueryWithOutputSettingsPushDownVisitor = QueryWithOutputSettingsPushDownMatcher::Visitor;
-
-}
--- a/src/Storages/MergeTree/MergeTreePrefetchedReadPool.cpp
+++ b/src/Storages/MergeTree/MergeTreePrefetchedReadPool.cpp
@ -395,6 +395,10 @@ void MergeTreePrefetchedReadPool::fillPerThreadTasks(size_t threads, size_t sum_

        part_stat.prefetch_step_marks = std::max(part_stat.prefetch_step_marks, per_part_infos[i]->min_marks_per_task);

+        if (part_stat.prefetch_step_marks == 0)
+            throw Exception(
+                ErrorCodes::BAD_ARGUMENTS, "Chosen number of marks to read is zero (likely because of weird interference of settings)");
+
        LOG_DEBUG(
            log,
            "Part: {}, sum_marks: {}, approx mark size: {}, prefetch_step_bytes: {}, prefetch_step_marks: {}, (ranges: {})",
--- a/src/Storages/MergeTree/MergeTreeReadPool.cpp
+++ b/src/Storages/MergeTree/MergeTreeReadPool.cpp
@ -24,6 +24,7 @@ namespace ErrorCodes
 {
 extern const int CANNOT_SCHEDULE_TASK;
 extern const int LOGICAL_ERROR;
+extern const int BAD_ARGUMENTS;
 }

 MergeTreeReadPool::MergeTreeReadPool(
@ -235,6 +236,10 @@ void MergeTreeReadPool::fillPerThreadInfo(size_t threads, size_t sum_marks)
            const auto part_idx = current_parts.back().part_idx;
            const auto min_marks_per_task = per_part_infos[part_idx]->min_marks_per_task;

+            if (min_marks_per_task == 0)
+                throw Exception(
+                    ErrorCodes::BAD_ARGUMENTS, "Chosen number of marks to read is zero (likely because of weird interference of settings)");
+
            /// Do not get too few rows from part.
            if (marks_in_part >= min_marks_per_task && need_marks < min_marks_per_task)
                need_marks = min_marks_per_task;
--- a/src/Storages/MergeTree/MergeTreeReadPoolBase.cpp
+++ b/src/Storages/MergeTree/MergeTreeReadPoolBase.cpp
@ -85,6 +85,7 @@ static size_t calculateMinMarksPerTask(
            min_marks_per_task = heuristic_min_marks;
        }
    }
+
    LOG_TEST(&Poco::Logger::get("MergeTreeReadPoolBase"), "Will use min_marks_per_task={}", min_marks_per_task);
    return min_marks_per_task;
 }
--- a/src/Storages/MergeTree/MergeTreeReadPoolParallelReplicas.cpp
+++ b/src/Storages/MergeTree/MergeTreeReadPoolParallelReplicas.cpp
@ -83,6 +83,7 @@ namespace DB
 namespace ErrorCodes
 {
    extern const int LOGICAL_ERROR;
+    extern const int BAD_ARGUMENTS;
 }

 MergeTreeReadPoolParallelReplicas::MergeTreeReadPoolParallelReplicas(
@ -120,6 +121,10 @@ MergeTreeReadPoolParallelReplicas::MergeTreeReadPoolParallelReplicas(
    for (const auto & info : per_part_infos)
        min_marks_per_task = std::max(min_marks_per_task, info->min_marks_per_task);

+    if (min_marks_per_task == 0)
+        throw Exception(
+            ErrorCodes::BAD_ARGUMENTS, "Chosen number of marks to read is zero (likely because of weird interference of settings)");
+
    extension.all_callback(InitialAllRangesAnnouncement(
        coordination_mode, parts_ranges.getDescriptions(), extension.number_of_current_replica, mark_segment_size));
 }
--- a/src/Storages/MergeTree/MergeTreeReadPoolParallelReplicasInOrder.cpp
+++ b/src/Storages/MergeTree/MergeTreeReadPoolParallelReplicasInOrder.cpp
@ -11,6 +11,7 @@ namespace DB
 namespace ErrorCodes
 {
    extern const int LOGICAL_ERROR;
+    extern const int BAD_ARGUMENTS;
 }

 MergeTreeReadPoolParallelReplicasInOrder::MergeTreeReadPoolParallelReplicasInOrder(
@ -42,6 +43,10 @@ MergeTreeReadPoolParallelReplicasInOrder::MergeTreeReadPoolParallelReplicasInOrd
    for (const auto & info : per_part_infos)
        min_marks_per_task = std::max(min_marks_per_task, info->min_marks_per_task);

+    if (min_marks_per_task == 0)
+        throw Exception(
+            ErrorCodes::BAD_ARGUMENTS, "Chosen number of marks to read is zero (likely because of weird interference of settings)");
+
    for (const auto & part : parts_ranges)
        request.push_back({part.data_part->info, MarkRanges{}});

--- a/src/Storages/MergeTree/ParallelReplicasReadingCoordinator.cpp
+++ b/src/Storages/MergeTree/ParallelReplicasReadingCoordinator.cpp
@ -1005,6 +1005,10 @@ void ParallelReplicasReadingCoordinator::handleInitialAllRangesAnnouncement(Init

 ParallelReadResponse ParallelReplicasReadingCoordinator::handleRequest(ParallelReadRequest request)
 {
+    if (request.min_number_of_marks == 0)
+        throw Exception(
+            ErrorCodes::BAD_ARGUMENTS, "Chosen number of marks to read is zero (likely because of weird interference of settings)");
+
    ProfileEventTimeIncrement<Microseconds> watch(ProfileEvents::ParallelReplicasHandleRequestMicroseconds);

    std::lock_guard lock(mutex);
--- a/src/Storages/MergeTree/ReplicatedMergeTreeRestartingThread.cpp
+++ b/src/Storages/MergeTree/ReplicatedMergeTreeRestartingThread.cpp
@ -7,7 +7,6 @@
 #include <Interpreters/Context.h>
 #include <Common/FailPoint.h>
 #include <Common/ZooKeeper/KeeperException.h>
-#include <Common/randomSeed.h>
 #include <Core/ServerUUID.h>
 #include <boost/algorithm/string/replace.hpp>

@ -49,6 +48,20 @@ ReplicatedMergeTreeRestartingThread::ReplicatedMergeTreeRestartingThread(Storage
    task = storage.getContext()->getSchedulePool().createTask(log_name, [this]{ run(); });
 }

+void ReplicatedMergeTreeRestartingThread::start(bool schedule)
+{
+    LOG_TRACE(log, "Starting the restating thread, schedule: {}", schedule);
+    if (schedule)
+        task->activateAndSchedule();
+    else
+        task->activate();
+}
+
+void ReplicatedMergeTreeRestartingThread::wakeup()
+{
+    task->schedule();
+}
+
 void ReplicatedMergeTreeRestartingThread::run()
 {
    if (need_stop)
--- a/src/Storages/MergeTree/ReplicatedMergeTreeRestartingThread.h
+++ b/src/Storages/MergeTree/ReplicatedMergeTreeRestartingThread.h
@ -24,16 +24,9 @@ class ReplicatedMergeTreeRestartingThread
 public:
    explicit ReplicatedMergeTreeRestartingThread(StorageReplicatedMergeTree & storage_);

-    void start(bool schedule = true)
-    {
-        LOG_TRACE(log, "Starting restating thread, schedule: {}", schedule);
-        if (schedule)
-            task->activateAndSchedule();
-        else
-            task->activate();
-    }
+    void start(bool schedule);

-    void wakeup() { task->schedule(); }
+    void wakeup();

    void shutdown(bool part_of_full_shutdown);

--- a/src/Storages/StorageReplicatedMergeTree.cpp
+++ b/src/Storages/StorageReplicatedMergeTree.cpp
@ -5194,17 +5194,16 @@ void StorageReplicatedMergeTree::startupImpl(bool from_attach_thread)

        startBeingLeader();

-        /// Activate replica in a separate thread if we are not calling from attach thread
-        restarting_thread.start(/*schedule=*/!from_attach_thread);
-
        if (from_attach_thread)
        {
            LOG_TRACE(log, "Trying to startup table from right now");
-            /// Try activating replica in current thread.
+            /// Try activating replica in the current thread.
            restarting_thread.run();
+            restarting_thread.start(false);
        }
        else
        {
+            restarting_thread.start(true);
            /// Wait while restarting_thread finishing initialization.
            /// NOTE It does not mean that replication is actually started after receiving this event.
            /// It only means that an attempt to startup replication was made.
@ -5225,7 +5224,7 @@ void StorageReplicatedMergeTree::startupImpl(bool from_attach_thread)
        session_expired_callback_handler = EventNotifier::instance().subscribe(Coordination::Error::ZSESSIONEXPIRED, [this]()
        {
            LOG_TEST(log, "Received event for expired session. Waking up restarting thread");
-            restarting_thread.start();
+            restarting_thread.start(true);
        });

        startBackgroundMovesIfNeeded();
@ -5294,7 +5293,6 @@ void StorageReplicatedMergeTree::flushAndPrepareForShutdown()
            LOG_TRACE(log, "The attach thread is shutdown");
        }

-
        restarting_thread.shutdown(/* part_of_full_shutdown */true);
        /// Explicitly set the event, because the restarting thread will not set it again
        startup_event.set();
@ -6342,7 +6340,7 @@ void StorageReplicatedMergeTree::alter(
                                "Metadata on replica is not up to date with common metadata in Zookeeper. "
                                "It means that this replica still not applied some of previous alters."
                                " Probably too many alters executing concurrently (highly not recommended). "
-                                "You can retry this error");
+                                "You can retry the query");

            /// Cannot retry automatically, because some zookeeper ops were lost on the first attempt. Will retry on DDLWorker-level.
            if (query_context->getZooKeeperMetadataTransaction())
--- a/src/Storages/System/StorageSystemOne.cpp
+++ b/src/Storages/System/StorageSystemOne.cpp
@ -41,7 +41,10 @@ Pipe StorageSystemOne::read(
    auto column = DataTypeUInt8().createColumnConst(1, 0u)->convertToFullColumnIfConst();
    Chunk chunk({ std::move(column) }, 1);

-    return Pipe(std::make_shared<SourceFromSingleChunk>(std::move(header), std::move(chunk)));
+    auto source = std::make_shared<SourceFromSingleChunk>(std::move(header), std::move(chunk));
+    source->addTotalRowsApprox(1);
+
+    return Pipe(source);
 }


--- a/tests/clickhouse-test
+++ b/tests/clickhouse-test
@ -59,6 +59,7 @@ MESSAGES_TO_RETRY = [
    "is already started to be removing by another replica right now",
    # This is from LSan, and it indicates its own internal problem:
    "Unable to get registers from thread",
+    "You can retry",
 ]

 MAX_RETRIES = 3
--- a/tests/config/install.sh
+++ b/tests/config/install.sh
@ -94,6 +94,7 @@ ln -sf $SRC_PATH/users.d/prefetch_settings.xml $DEST_SERVER_PATH/users.d/
 ln -sf $SRC_PATH/users.d/nonconst_timezone.xml $DEST_SERVER_PATH/users.d/
 ln -sf $SRC_PATH/users.d/allow_introspection_functions.yaml $DEST_SERVER_PATH/users.d/
 ln -sf $SRC_PATH/users.d/replicated_ddl_entry.xml $DEST_SERVER_PATH/users.d/
+ln -sf $SRC_PATH/users.d/limits.yaml $DEST_SERVER_PATH/users.d/

 if [[ -n "$USE_OLD_ANALYZER" ]] && [[ "$USE_OLD_ANALYZER" -eq 1 ]]; then
    ln -sf $SRC_PATH/users.d/analyzer.xml $DEST_SERVER_PATH/users.d/
--- a/tests/config/users.d/limits.yaml
+++ b/tests/config/users.d/limits.yaml
@ -0,0 +1,56 @@
+profiles:
+  default:
+    max_memory_usage: 5G
+    max_rows_to_read: 20000000
+
+    # Also set every other limit to a high value, so it will not limit anything, but we will test that code around it.
+    s3_max_get_rps: 1000000
+    s3_max_get_burst: 2000000
+    s3_max_put_rps: 1000000
+    s3_max_put_burst: 2000000
+    max_remote_read_network_bandwidth: 1T
+    max_remote_write_network_bandwidth: 1T
+    max_local_read_bandwidth: 1T
+    max_local_write_bandwidth: 1T
+    use_index_for_in_with_subqueries_max_values: 1G
+    max_bytes_to_read: 1T
+    max_bytes_to_read_leaf: 1T
+    max_rows_to_group_by: 10G
+    max_bytes_before_external_group_by: 10G
+    max_rows_to_sort: 10G
+    max_bytes_to_sort: 10G
+    max_bytes_before_external_sort: 10G
+    max_result_rows: 1G
+    max_result_bytes: 1G
+    max_execution_time: 600
+    max_execution_time_leaf: 600
+    max_execution_speed: 100G
+    max_execution_speed_bytes: 10T
+    timeout_before_checking_execution_speed: 300
+    max_estimated_execution_time: 600
+    max_columns_to_read: 20K
+    max_temporary_columns: 20K
+    max_temporary_non_const_columns: 20K
+    max_rows_in_set: 10G
+    max_bytes_in_set: 10G
+    max_rows_in_join: 10G
+    max_bytes_in_join: 10G
+    max_rows_to_transfer: 1G
+    max_bytes_to_transfer: 1G
+    max_rows_in_distinct: 10G
+    max_bytes_in_distinct: 10G
+    max_memory_usage_for_user: 32G
+    max_network_bandwidth: 100G
+    max_network_bytes: 1T
+    max_network_bandwidth_for_user: 100G
+    max_network_bandwidth_for_all_users: 100G
+    max_temporary_data_on_disk_size_for_user: 100G
+    max_temporary_data_on_disk_size_for_query: 100G
+    max_backup_bandwidth: 100G
+    max_hyperscan_regexp_length: 1M
+    max_hyperscan_regexp_total_length: 10M
+    query_cache_max_size_in_bytes: 10M
+    query_cache_max_entries: 100K
+    external_storage_max_read_rows: 10G
+    external_storage_max_read_bytes: 10G
+    max_streams_for_merge_tree_reading: 1000
--- a/tests/queries/0_stateless/00002_log_and_exception_messages_formatting.sql
+++ b/tests/queries/0_stateless/00002_log_and_exception_messages_formatting.sql
@ -9,6 +9,8 @@ system flush logs;
 drop table if exists logs;
 create view logs as select * from system.text_log where now() - toIntervalMinute(120) < event_time;

+SET max_rows_to_read = 0;
+
 -- Check that we don't have too many messages formatted with fmt::runtime or strings concatenation.
 -- 0.001 threshold should be always enough, the value was about 0.00025
 WITH 0.001 AS threshold
--- a/tests/queries/0_stateless/00086_concat_nary_const_with_nonconst_segfault.sql
+++ b/tests/queries/0_stateless/00086_concat_nary_const_with_nonconst_segfault.sql
@ -1 +1 @@
-SELECT extract(toString(number), '10000000') FROM system.numbers_mt WHERE concat(materialize('1'), '...', toString(number)) LIKE '%10000000%' LIMIT 1
+SELECT extract(toString(number), '10000000') FROM system.numbers_mt WHERE concat(materialize('1'), '...', toString(number)) LIKE '%10000000%' LIMIT 1 SETTINGS max_rows_to_read = 0;
--- a/tests/queries/0_stateless/00111_shard_external_sort_distributed.reference
+++ b/tests/queries/0_stateless/00111_shard_external_sort_distributed.reference
@ -1,20 +1,20 @@
-7040546
-7040546
-4327029
-4327029
-1613512
-1613512
-8947307
-8947307
-6233790
-6233790
-3520273
-3520273
-806756
-806756
-8140551
-8140551
-5427034
-5427034
-2713517
-2713517
+4437158
+4437158
+1723641
+1723641
+3630402
+3630402
+916885
+916885
+2823646
+2823646
+110129
+110129
+4730407
+4730407
+2016890
+2016890
+3923651
+3923651
+1210134
+1210134
--- a/tests/queries/0_stateless/00111_shard_external_sort_distributed.sql
+++ b/tests/queries/0_stateless/00111_shard_external_sort_distributed.sql
@ -1,11 +1,12 @@
-- Tags: distributed
+-- Tags: distributed, long, no-flaky-check
+-- ^ no-flaky-check - sometimes longer than 600s with ThreadFuzzer.

-SET max_memory_usage = 300000000;
-SET max_bytes_before_external_sort = 20000000;
+SET max_memory_usage = 150000000;
+SET max_bytes_before_external_sort = 10000000;

 DROP TABLE IF EXISTS numbers10m;
-CREATE VIEW numbers10m AS SELECT number FROM system.numbers LIMIT 10000000;
+CREATE VIEW numbers10m AS SELECT number FROM system.numbers LIMIT 5000000;

-SELECT number FROM remote('127.0.0.{2,3}', currentDatabase(), numbers10m) ORDER BY number * 1234567890123456789 LIMIT 19999980, 20;
+SELECT number FROM remote('127.0.0.{2,3}', currentDatabase(), numbers10m) ORDER BY number * 1234567890123456789 LIMIT 4999980, 20;

 DROP TABLE numbers10m;
--- a/tests/queries/0_stateless/00375_shard_group_uniq_array_of_string.sql
+++ b/tests/queries/0_stateless/00375_shard_group_uniq_array_of_string.sql
@ -1,4 +1,4 @@
-- Tags: shard
+-- Tags: shard, long

 DROP TABLE IF EXISTS group_uniq_str;
 CREATE TABLE group_uniq_str ENGINE = Memory AS SELECT number % 10 as id, toString(intDiv((number%10000), 10)) as v FROM system.numbers LIMIT 10000000;
@ -7,7 +7,7 @@ INSERT INTO group_uniq_str SELECT 2 as id, toString(number % 100) as v FROM syst
 INSERT INTO group_uniq_str SELECT 5 as id, toString(number % 100) as v FROM system.numbers LIMIT 10000000;

 SELECT length(groupUniqArray(v)) FROM group_uniq_str GROUP BY id ORDER BY id;
-SELECT length(groupUniqArray(v)) FROM remote('127.0.0.{2,3,4,5}', currentDatabase(), 'group_uniq_str') GROUP BY id ORDER BY id;
+SELECT length(groupUniqArray(v)) FROM remote('127.0.0.{2,3,4,5}', currentDatabase(), 'group_uniq_str') GROUP BY id ORDER BY id SETTINGS max_rows_to_read = '100M';
 SELECT length(groupUniqArray(10)(v)) FROM group_uniq_str GROUP BY id ORDER BY id;
 SELECT length(groupUniqArray(10000)(v)) FROM group_uniq_str GROUP BY id ORDER BY id;

--- a/tests/queries/0_stateless/00376_shard_group_uniq_array_of_int_array.sql
+++ b/tests/queries/0_stateless/00376_shard_group_uniq_array_of_int_array.sql
@ -1,4 +1,6 @@
-- Tags: shard
+-- Tags: long
+
+SET max_rows_to_read = '55M';

 DROP TABLE IF EXISTS group_uniq_arr_int;
 CREATE TABLE group_uniq_arr_int ENGINE = Memory AS
--- a/tests/queries/0_stateless/00377_shard_group_uniq_array_of_string_array.sql
+++ b/tests/queries/0_stateless/00377_shard_group_uniq_array_of_string_array.sql
@ -1,4 +1,5 @@
-- Tags: shard
+-- Tags: shard, long
+SET max_rows_to_read = '55M';

 DROP TABLE IF EXISTS group_uniq_arr_str;
 CREATE TABLE group_uniq_arr_str ENGINE = Memory AS
--- a/tests/queries/0_stateless/00463_long_sessions_in_http_interface.sh
+++ b/tests/queries/0_stateless/00463_long_sessions_in_http_interface.sh
@ -74,7 +74,7 @@ ${CLICKHOUSE_CLIENT} --query "DROP TABLE t"

 echo "A session cannot be used by concurrent connections:"

-${CLICKHOUSE_CURL} -sS -X POST "${CLICKHOUSE_URL}&session_id=${CLICKHOUSE_DATABASE}_9&query_id=${CLICKHOUSE_DATABASE}_9" --data-binary "SELECT count() FROM system.numbers" >/dev/null &
+${CLICKHOUSE_CURL} -sS -X POST "${CLICKHOUSE_URL}&session_id=${CLICKHOUSE_DATABASE}_9&query_id=${CLICKHOUSE_DATABASE}_9&max_rows_to_read=0" --data-binary "SELECT count() FROM system.numbers" >/dev/null &

 # An infinite loop is required to make the test reliable. We will ensure that at least once the query on the line above has started before this check
 while true
--- a/tests/queries/0_stateless/00600_replace_running_query.sh
+++ b/tests/queries/0_stateless/00600_replace_running_query.sh
@ -8,7 +8,7 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)

 TEST_PREFIX="${CLICKHOUSE_DATABASE}"
 ${CLICKHOUSE_CLIENT} -q "drop user if exists u_00600${TEST_PREFIX}"
-${CLICKHOUSE_CLIENT} -q "create user u_00600${TEST_PREFIX} settings max_execution_time=60, readonly=1"
+${CLICKHOUSE_CLIENT} -q "create user u_00600${TEST_PREFIX} settings max_execution_time=60, readonly=1, max_rows_to_read=0"
 ${CLICKHOUSE_CLIENT} -q "grant select on system.numbers to u_00600${TEST_PREFIX}"

 function wait_for_query_to_start()
@ -28,7 +28,7 @@ function wait_for_queries_to_finish()
 }


-$CLICKHOUSE_CURL -sS "$CLICKHOUSE_URL&query_id=${CLICKHOUSE_DATABASE}hello&replace_running_query=1" -d 'SELECT 1, count() FROM system.numbers' > /dev/null 2>&1 &
+$CLICKHOUSE_CURL -sS "$CLICKHOUSE_URL&query_id=${CLICKHOUSE_DATABASE}hello&replace_running_query=1&max_rows_to_read=0" -d 'SELECT 1, count() FROM system.numbers' > /dev/null 2>&1 &
 wait_for_query_to_start "${CLICKHOUSE_DATABASE}hello"

 # Replace it
@ -51,7 +51,7 @@ $CLICKHOUSE_CURL -sS "$CLICKHOUSE_URL" -d "KILL QUERY WHERE query_id = '${CLICKH
 wait
 wait_for_queries_to_finish

-${CLICKHOUSE_CLIENT} --query_id="${CLICKHOUSE_DATABASE}42" --query='SELECT 3, count() FROM system.numbers' 2>&1 | grep -cF 'QUERY_WAS_CANCELLED' &
+${CLICKHOUSE_CLIENT} --query_id="${CLICKHOUSE_DATABASE}42" --max_rows_to_read=0 --query='SELECT 3, count() FROM system.numbers' 2>&1 | grep -cF 'QUERY_WAS_CANCELLED' &
 wait_for_query_to_start "${CLICKHOUSE_DATABASE}42"
 ${CLICKHOUSE_CLIENT} --query_id="${CLICKHOUSE_DATABASE}42" --replace_running_query=1 --replace_running_query_max_wait_ms=500 --query='SELECT 43' 2>&1 | grep -F "can't be stopped" > /dev/null
 wait
--- a/tests/queries/0_stateless/00601_kill_running_query.reference
+++ b/tests/queries/0_stateless/00601_kill_running_query.reference
@ -1 +1 @@
-waiting	test_00601_default	default	SELECT sum(ignore(*)) FROM (SELECT number % 1000 AS k, groupArray(number) FROM numbers(50000000) GROUP BY k)
+waiting	test_00601_default	default	SELECT sum(ignore(*)) FROM (SELECT number % 1000 AS k, groupArray(number) FROM numbers(50000000) GROUP BY k) SETTINGS max_rows_to_read = 0
--- a/tests/queries/0_stateless/00601_kill_running_query.sh
+++ b/tests/queries/0_stateless/00601_kill_running_query.sh
@ -11,7 +11,7 @@ function wait_for_query_to_start()
    while [[ $($CLICKHOUSE_CURL -sS "$CLICKHOUSE_URL" -d "SELECT count() FROM system.processes WHERE query_id = '$1'") == 0 ]]; do sleep 0.1; done
 }

-${CLICKHOUSE_CURL_COMMAND} -q --max-time 30 -sS "$CLICKHOUSE_URL&query_id=test_00601_$CLICKHOUSE_DATABASE" -d 'SELECT sum(ignore(*)) FROM (SELECT number % 1000 AS k, groupArray(number) FROM numbers(50000000) GROUP BY k)' > /dev/null &
+${CLICKHOUSE_CURL_COMMAND} -q --max-time 30 -sS "$CLICKHOUSE_URL&query_id=test_00601_$CLICKHOUSE_DATABASE" -d 'SELECT sum(ignore(*)) FROM (SELECT number % 1000 AS k, groupArray(number) FROM numbers(50000000) GROUP BY k) SETTINGS max_rows_to_read = 0' > /dev/null &
 wait_for_query_to_start "test_00601_$CLICKHOUSE_DATABASE"
 $CLICKHOUSE_CURL -sS "$CLICKHOUSE_URL" -d "KILL QUERY WHERE query_id = 'test_00601_$CLICKHOUSE_DATABASE'"
 wait
--- a/tests/queries/0_stateless/00632_get_sample_block_cache.sql
+++ b/tests/queries/0_stateless/00632_get_sample_block_cache.sql
--- a/tests/queries/0_stateless/00804_test_deflate_qpl_codec_compression.reference
+++ b/tests/queries/0_stateless/00804_test_deflate_qpl_codec_compression.reference
@ -1,4 +1,4 @@
-CREATE TABLE default.compression_codec\n(\n    `id` UInt64 CODEC(DEFLATE_QPL),\n    `data` String CODEC(DEFLATE_QPL),\n    `ddd` Date CODEC(DEFLATE_QPL),\n    `ddd32` Date32 CODEC(DEFLATE_QPL),\n    `somenum` Float64 CODEC(DEFLATE_QPL),\n    `somestr` FixedString(3) CODEC(DEFLATE_QPL),\n    `othernum` Int64 CODEC(DEFLATE_QPL),\n    `somearray` Array(UInt8) CODEC(DEFLATE_QPL),\n    `somemap` Map(String, UInt32) CODEC(DEFLATE_QPL),\n    `sometuple` Tuple(UInt16, UInt64) CODEC(DEFLATE_QPL)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS index_granularity = 8192
+CREATE TABLE default.compression_codec\n(\n    `id` UInt64 CODEC(DEFLATE_QPL),\n    `data` String CODEC(DEFLATE_QPL),\n    `ddd` Date CODEC(DEFLATE_QPL),\n    `ddd32` Date32 CODEC(DEFLATE_QPL),\n    `somenum` Float64 CODEC(DEFLATE_QPL),\n    `somestr` FixedString(3) CODEC(DEFLATE_QPL),\n    `othernum` Int64 CODEC(DEFLATE_QPL),\n    `somearray` Array(UInt8) CODEC(DEFLATE_QPL),\n    `somemap` Map(String, UInt32) CODEC(DEFLATE_QPL),\n    `sometuple` Tuple(\n        UInt16,\n        UInt64) CODEC(DEFLATE_QPL)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS index_granularity = 8192
 1	hello	2018-12-14	2018-12-14	1.1	aaa	5	[1,2,3]	{'k1':1,'k2':2}	(1,2)
 2	world	2018-12-15	2018-12-15	2.2	bbb	6	[4,5,6]	{'k3':3,'k4':4}	(3,4)
 3	!	2018-12-16	2018-12-16	3.3	ccc	7	[7,8,9]	{'k5':5,'k6':6}	(5,6)
--- a/tests/queries/0_stateless/00804_test_zstd_qat_codec_compression.reference
+++ b/tests/queries/0_stateless/00804_test_zstd_qat_codec_compression.reference
@ -1,4 +1,4 @@
-CREATE TABLE default.compression_codec\n(\n    `id` UInt64 CODEC(ZSTD_QAT(1)),\n    `data` String CODEC(ZSTD_QAT(1)),\n    `ddd` Date CODEC(ZSTD_QAT(1)),\n    `ddd32` Date32 CODEC(ZSTD_QAT(1)),\n    `somenum` Float64 CODEC(ZSTD_QAT(1)),\n    `somestr` FixedString(3) CODEC(ZSTD_QAT(1)),\n    `othernum` Int64 CODEC(ZSTD_QAT(1)),\n    `somearray` Array(UInt8) CODEC(ZSTD_QAT(1)),\n    `somemap` Map(String, UInt32) CODEC(ZSTD_QAT(1)),\n    `sometuple` Tuple(UInt16, UInt64) CODEC(ZSTD_QAT(1))\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS index_granularity = 8192
+CREATE TABLE default.compression_codec\n(\n    `id` UInt64 CODEC(ZSTD_QAT(1)),\n    `data` String CODEC(ZSTD_QAT(1)),\n    `ddd` Date CODEC(ZSTD_QAT(1)),\n    `ddd32` Date32 CODEC(ZSTD_QAT(1)),\n    `somenum` Float64 CODEC(ZSTD_QAT(1)),\n    `somestr` FixedString(3) CODEC(ZSTD_QAT(1)),\n    `othernum` Int64 CODEC(ZSTD_QAT(1)),\n    `somearray` Array(UInt8) CODEC(ZSTD_QAT(1)),\n    `somemap` Map(String, UInt32) CODEC(ZSTD_QAT(1)),\n    `sometuple` Tuple(\n        UInt16,\n        UInt64) CODEC(ZSTD_QAT(1))\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS index_granularity = 8192
 1	hello	2018-12-14	2018-12-14	1.1	aaa	5	[1,2,3]	{'k1':1,'k2':2}	(1,2)
 2	world	2018-12-15	2018-12-15	2.2	bbb	6	[4,5,6]	{'k3':3,'k4':4}	(3,4)
 3	!	2018-12-16	2018-12-16	3.3	ccc	7	[7,8,9]	{'k5':5,'k6':6}	(5,6)
--- a/tests/queries/0_stateless/00834_cancel_http_readonly_queries_on_client_close.sh
+++ b/tests/queries/0_stateless/00834_cancel_http_readonly_queries_on_client_close.sh
@ -5,7 +5,7 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
 . "$CURDIR"/../shell_config.sh

-${CLICKHOUSE_CURL} --max-time 1 -sS "${CLICKHOUSE_URL}&query_id=cancel_http_readonly_queries_on_client_close&cancel_http_readonly_queries_on_client_close=1&query=SELECT+count()+FROM+system.numbers" 2>&1 | grep -cF 'curl: (28)'
+${CLICKHOUSE_CURL} --max-time 1 -sS "${CLICKHOUSE_URL}&query_id=cancel_http_readonly_queries_on_client_close&cancel_http_readonly_queries_on_client_close=1&max_rows_to_read=0&query=SELECT+count()+FROM+system.numbers" 2>&1 | grep -cF 'curl: (28)'

 i=0 retries=300
 while [[ $i -lt $retries ]]; do
--- a/tests/queries/0_stateless/00857_global_joinsavel_table_alias.sql
+++ b/tests/queries/0_stateless/00857_global_joinsavel_table_alias.sql
@ -1,4 +1,3 @@
-
 DROP TABLE IF EXISTS local_table;
 DROP TABLE IF EXISTS other_table;

--- a/tests/queries/0_stateless/00906_low_cardinality_cache.sql
+++ b/tests/queries/0_stateless/00906_low_cardinality_cache.sql
@ -1,3 +1,6 @@
+-- Tags: long
+
+SET max_rows_to_read = '100M';
 drop table if exists lc_00906;
 create table lc_00906 (b LowCardinality(String)) engine=MergeTree order by b SETTINGS index_granularity = 8192, index_granularity_bytes = '10Mi';
 insert into lc_00906 select '0123456789' from numbers(100000000);
--- a/tests/queries/0_stateless/00974_query_profiler.sql
+++ b/tests/queries/0_stateless/00974_query_profiler.sql
@ -17,6 +17,7 @@ SELECT count() > 0 FROM system.trace_log t WHERE query_id = (SELECT query_id FRO
 SET query_profiler_real_time_period_ns = 0;
 SET query_profiler_cpu_time_period_ns = 1000000;
 SET log_queries = 1;
+SET max_rows_to_read = 0;
 SELECT count(), ignore('test cpu time query profiler') FROM numbers_mt(10000000000);
 SET log_queries = 0;
 SYSTEM FLUSH LOGS;
--- a/tests/queries/0_stateless/00976_max_execution_speed.sql
+++ b/tests/queries/0_stateless/00976_max_execution_speed.sql
@ -1,2 +1,2 @@
-SET max_execution_speed = 1, max_execution_time = 3;
+SET max_execution_speed = 1, max_execution_time = 3, max_rows_to_read = 0;
 SELECT count() FROM system.numbers; -- { serverError TIMEOUT_EXCEEDED }
--- a/tests/queries/0_stateless/01010_pmj_right_table_memory_limits.sql
+++ b/tests/queries/0_stateless/01010_pmj_right_table_memory_limits.sql
@ -1,5 +1,7 @@
 -- Tags: no-parallel, no-fasttest, no-random-settings

+SET max_bytes_in_join = 0;
+SET max_rows_in_join = 0;
 SET max_memory_usage = 32000000;
 SET join_on_disk_max_files_to_merge = 4;

--- a/tests/queries/0_stateless/01017_uniqCombined_memory_usage.sql
+++ b/tests/queries/0_stateless/01017_uniqCombined_memory_usage.sql
@ -7,7 +7,8 @@
 -- sizeof(HLL) is (2^K * 6 / 8)
 -- hence max_memory_usage for 100 rows = (96<<10)*100 = 9830400

-SET use_uncompressed_cache = 0; 
+SET use_uncompressed_cache = 0;
+SET memory_profiler_step = 1;

 -- HashTable for UInt32 (used until (1<<13) elements), hence 8192 elements
 SELECT 'UInt32';
@ -31,14 +32,14 @@ SELECT 'K=16';
 SELECT 'UInt32';
 SET max_memory_usage = 2000000;
 SELECT sum(u) FROM (SELECT intDiv(number, 4096) AS k, uniqCombined(16)(number % 4096) u FROM numbers(4096 * 100) GROUP BY k); -- { serverError MEMORY_LIMIT_EXCEEDED }
-SET max_memory_usage = 4915200;
+SET max_memory_usage = 5230000;
 SELECT sum(u) FROM (SELECT intDiv(number, 4096) AS k, uniqCombined(16)(number % 4096) u FROM numbers(4096 * 100) GROUP BY k);

 -- HashTable for UInt64 (used until (1<<11) elements), hence 2048 elements
 SELECT 'UInt64';
 SET max_memory_usage = 2000000;
 SELECT sum(u) FROM (SELECT intDiv(number, 2048) AS k, uniqCombined(16)(reinterpretAsString(number % 2048)) u FROM numbers(2048 * 100) GROUP BY k); -- { serverError MEMORY_LIMIT_EXCEEDED }
-SET max_memory_usage = 4915200;
+SET max_memory_usage = 5900000;
 SELECT sum(u) FROM (SELECT intDiv(number, 2048) AS k, uniqCombined(16)(reinterpretAsString(number % 2048)) u FROM numbers(2048 * 100) GROUP BY k);

 SELECT 'K=18';
--- a/tests/queries/0_stateless/01091_query_profiler_does_not_hang.sql
+++ b/tests/queries/0_stateless/01091_query_profiler_does_not_hang.sql
@ -1,4 +1,4 @@
 -- Tags: no-tsan, no-asan, no-ubsan, no-msan, no-debug

-SET query_profiler_cpu_time_period_ns = 1;
+SET query_profiler_cpu_time_period_ns = 1, max_rows_to_read = 0;
 SELECT count() FROM numbers_mt(1000000000);
--- a/tests/queries/0_stateless/01119_optimize_trivial_insert_select.sql
+++ b/tests/queries/0_stateless/01119_optimize_trivial_insert_select.sql
@ -1,8 +1,9 @@
 drop table if exists t;
 create table t(n int, a Int64, s String) engine = MergeTree() order by a;

-set enable_positional_arguments=0;
-set optimize_trivial_insert_select=1;
+set enable_positional_arguments = 0;
+set optimize_trivial_insert_select = 1;
+set max_rows_to_read = 0;

 -- due to aggregate functions, optimize_trivial_insert_select will not be applied
 insert into t select 1, sum(number) as c, getSetting('max_threads') from numbers_mt(100000000) settings max_insert_threads=4, max_threads=2;
--- a/tests/queries/0_stateless/01161_all_system_tables.sh
+++ b/tests/queries/0_stateless/01161_all_system_tables.sh
@ -19,7 +19,7 @@ function run_selects()
    thread_num=$1
    readarray -t tables_arr < <(${CLICKHOUSE_CLIENT} -q "SELECT database || '.' || name FROM system.tables
    WHERE database in ('system', 'information_schema', 'INFORMATION_SCHEMA') and name != 'zookeeper' and name != 'models'
-    AND sipHash64(name || toString($RAND)) % $THREADS = $thread_num AND name NOT LIKE '%\\_sender' AND name NOT LIKE '%\\_watcher'")
+    AND sipHash64(name || toString($RAND)) % $THREADS = $thread_num AND name NOT LIKE '%\\_sender' AND name NOT LIKE '%\\_watcher' AND name != 'coverage_log'")

    for t in "${tables_arr[@]}"
    do
--- a/tests/queries/0_stateless/01191_rename_dictionary.sql
+++ b/tests/queries/0_stateless/01191_rename_dictionary.sql
@ -27,6 +27,7 @@ RENAME DICTIONARY test_01191.t TO test_01191.dict1; -- {serverError INCORRECT_QU
 DROP DICTIONARY test_01191.t; -- {serverError INCORRECT_QUERY}
 DROP TABLE test_01191.t;

+DROP DATABASE IF EXISTS dummy_db;
 CREATE DATABASE dummy_db ENGINE=Atomic;
 RENAME DICTIONARY test_01191.dict TO dummy_db.dict1;
 RENAME DICTIONARY dummy_db.dict1 TO test_01191.dict;
--- a/tests/queries/0_stateless/01245_limit_infinite_sources.sql
+++ b/tests/queries/0_stateless/01245_limit_infinite_sources.sql
@ -9,3 +9,4 @@ FROM
 )
 WHERE number = 1
 LIMIT 1
+SETTINGS max_rows_to_read = 0;
--- a/tests/queries/0_stateless/01249_flush_interactive.sh
+++ b/tests/queries/0_stateless/01249_flush_interactive.sh
@ -14,10 +14,10 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)

 function test()
 {
-    timeout 5 ${CLICKHOUSE_LOCAL} --max_execution_time 10 --query "
+    timeout 5 ${CLICKHOUSE_LOCAL} --max_execution_time 10 --max_rows_to_read 0 --query "
        SELECT DISTINCT number % 5 FROM system.numbers" ||:
    echo -e '---'
-    timeout 5 ${CLICKHOUSE_CURL} -sS --no-buffer "${CLICKHOUSE_URL}&max_execution_time=10" --data-binary "
+    timeout 5 ${CLICKHOUSE_CURL} -sS --no-buffer "${CLICKHOUSE_URL}&max_execution_time=10&max_rows_to_read=0" --data-binary "
        SELECT DISTINCT number % 5 FROM system.numbers" ||:
    echo -e '---'
 }
--- a/tests/queries/0_stateless/01293_show_settings.reference
+++ b/tests/queries/0_stateless/01293_show_settings.reference
@ -5,5 +5,7 @@ connect_timeout_with_failover_secure_ms	Milliseconds	3000
 external_storage_connect_timeout_sec	UInt64	10
 s3_connect_timeout_ms	UInt64	1000
 filesystem_prefetch_max_memory_usage	UInt64	1073741824
+max_memory_usage	UInt64	5000000000
+max_memory_usage_for_user	UInt64	32000000000
 max_untracked_memory	UInt64	1048576
 memory_profiler_step	UInt64	1048576
--- a/tests/queries/0_stateless/01301_aggregate_state_exception_memory_leak.reference
+++ b/tests/queries/0_stateless/01301_aggregate_state_exception_memory_leak.reference
@ -1,2 +1,2 @@
-Memory limit (for query) exceeded
+Memory limit exceeded
 Ok
--- a/tests/queries/0_stateless/01301_aggregate_state_exception_memory_leak.sh
+++ b/tests/queries/0_stateless/01301_aggregate_state_exception_memory_leak.sh
@ -8,7 +8,7 @@ CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 start=$SECONDS
 # If the memory leak exists, it will lead to OOM fairly quickly.
 for _ in {1..1000}; do
-    $CLICKHOUSE_CLIENT --max_memory_usage 1G <<< "SELECT uniqExactState(number) FROM system.numbers_mt GROUP BY number % 10";
+    $CLICKHOUSE_CLIENT --max_memory_usage 1G --max_rows_to_read 0 <<< "SELECT uniqExactState(number) FROM system.numbers_mt GROUP BY number % 10";

    # NOTE: we cannot use timeout here since this will not guarantee that the query will be executed at least once.
    # (since graceful wait of clickhouse-client had been reverted)
@ -16,5 +16,5 @@ for _ in {1..1000}; do
    if [[ $elapsed -gt 30 ]]; then
        break
    fi
-done 2>&1 | grep -o -F 'Memory limit (for query) exceeded' | uniq
+done 2>&1 | grep -o -P 'Memory limit .+ exceeded' | sed -r -e 's/(Memory limit)(.+)( exceeded)/\1\3/' | uniq
 echo 'Ok'
--- a/tests/queries/0_stateless/01304_direct_io_long.sh
+++ b/tests/queries/0_stateless/01304_direct_io_long.sh
@ -1,18 +1,19 @@
 #!/usr/bin/env bash
-# Tags: long, no-object-storage-with-slow-build
+# Tags: long, no-object-storage-with-slow-build, no-flaky-check
+# It can be too long with ThreadFuzzer

 CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
 . "$CURDIR"/../shell_config.sh

-$CLICKHOUSE_CLIENT --query "
+$CLICKHOUSE_CLIENT --max_rows_to_read 50M --query "
    DROP TABLE IF EXISTS bug;
    CREATE TABLE bug (UserID UInt64, Date Date) ENGINE = MergeTree ORDER BY Date
        SETTINGS index_granularity = 8192, index_granularity_bytes = '10Mi', merge_max_block_size = 8192;
    INSERT INTO bug SELECT rand64(), '2020-06-07' FROM numbers(50000000);
    OPTIMIZE TABLE bug FINAL;"
 LOG="$CLICKHOUSE_TMP/err-$CLICKHOUSE_DATABASE"
-$CLICKHOUSE_BENCHMARK --iterations 10 --max_threads 100 --min_bytes_to_use_direct_io 1 <<< "SELECT sum(UserID) FROM bug PREWHERE NOT ignore(Date)" 1>/dev/null 2>"$LOG"
+$CLICKHOUSE_BENCHMARK --max_rows_to_read 51M --iterations 10 --max_threads 100 --min_bytes_to_use_direct_io 1 <<< "SELECT sum(UserID) FROM bug PREWHERE NOT ignore(Date)" 1>/dev/null 2>"$LOG"
 cat "$LOG" | grep Exception
 cat "$LOG" | grep Loaded

--- a/tests/queries/0_stateless/01401_FORMAT_SETTINGS.reference
+++ b/tests/queries/0_stateless/01401_FORMAT_SETTINGS.reference
@ -1,7 +1,7 @@
 1
 1
 1
-1
-1
+2
+1
 2
 2
--- a/tests/queries/0_stateless/01401_FORMAT_SETTINGS.sh
+++ b/tests/queries/0_stateless/01401_FORMAT_SETTINGS.sh
@ -13,7 +13,7 @@ ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" -d 'SELECT DISTINCT blockSize() FROM
 ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" -d 'SELECT DISTINCT blockSize() FROM numbers(2) FORMAT CSV SETTINGS max_block_size = 1'
 # push down append
 ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" -d 'SELECT DISTINCT blockSize() FROM numbers(2) SETTINGS max_compress_block_size = 1 FORMAT CSV SETTINGS max_block_size = 1'
-# overwrite on push down (since these settings goes latest)
+# not overwrite on push down
 ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" -d 'SELECT DISTINCT blockSize() FROM numbers(2) SETTINGS max_block_size = 2 FORMAT CSV SETTINGS max_block_size = 1'
 # on push-down
 ${CLICKHOUSE_CURL} -sS "${CLICKHOUSE_URL}" -d 'SELECT DISTINCT blockSize() FROM numbers(2) SETTINGS max_block_size = 1 FORMAT CSV'
--- a/tests/queries/0_stateless/01458_named_tuple_millin.reference
+++ b/tests/queries/0_stateless/01458_named_tuple_millin.reference
@ -1,12 +1,16 @@
 CREATE TABLE default.tuple
 (
-    `j` Tuple(a Int8, b String)
+    `j` Tuple(
+        a Int8,
+        b String)
 )
 ENGINE = Memory
 j	Tuple(\n    a Int8,\n    b String)					
 CREATE TABLE default.tuple
 (
-    `j` Tuple(a Int8, b String)
+    `j` Tuple(
+        a Int8,
+        b String)
 )
 ENGINE = Memory
 j	Tuple(\n    a Int8,\n    b String)					
--- a/tests/queries/0_stateless/01485_256_bit_multiply.sql
+++ b/tests/queries/0_stateless/01485_256_bit_multiply.sql
@ -1,5 +1,7 @@
 -- Tags: no-random-settings, no-asan, no-msan, no-tsan, no-ubsan, no-debug

+SET max_rows_to_read = '100M';
+
 select count() from
 (
    select toInt128(number) * number x, toInt256(number) * number y from numbers_mt(100000000) where x != y
--- a/tests/queries/0_stateless/01504_compression_multiple_streams.reference
+++ b/tests/queries/0_stateless/01504_compression_multiple_streams.reference
@ -1,20 +1,20 @@
 1	1	[[1]]	(1,[1])
 1	1	[[1]]	(1,[1])
-CREATE TABLE default.columns_with_multiple_streams\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(UInt32, Array(UInt64)) CODEC(T64, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 0, min_bytes_for_wide_part = 0, index_granularity = 8192
+CREATE TABLE default.columns_with_multiple_streams\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(\n        UInt32,\n        Array(UInt64)) CODEC(T64, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 0, min_bytes_for_wide_part = 0, index_granularity = 8192
 1	1	[[1]]	(1,[1])
 2	2	[[2]]	(2,[2])
-CREATE TABLE default.columns_with_multiple_streams\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(UInt32, Array(UInt64)) CODEC(Delta, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 0, min_bytes_for_wide_part = 0, index_granularity = 8192
+CREATE TABLE default.columns_with_multiple_streams\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(\n        UInt32,\n        Array(UInt64)) CODEC(Delta, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 0, min_bytes_for_wide_part = 0, index_granularity = 8192
 1	1	[[1]]	(1,[1])
 2	2	[[2]]	(2,[2])
 3	3	[[3]]	(3,[3])
 1	1	[[1]]	(1,[1])
 1	1	[[1]]	(1,[1])
-CREATE TABLE default.columns_with_multiple_streams_compact\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(UInt32, Array(UInt64)) CODEC(Delta, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 100000, min_bytes_for_wide_part = 100000, index_granularity = 8192
+CREATE TABLE default.columns_with_multiple_streams_compact\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(\n        UInt32,\n        Array(UInt64)) CODEC(Delta, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 100000, min_bytes_for_wide_part = 100000, index_granularity = 8192
 1	1	[[1]]	(1,[1])
 2	2	[[2]]	(2,[2])
 1	1	[[1]]	(1,[1])
 2	2	[[2]]	(2,[2])
-CREATE TABLE default.columns_with_multiple_streams_compact\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(UInt32, Array(UInt64)) CODEC(Delta, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 100000, min_bytes_for_wide_part = 100000, index_granularity = 8192
+CREATE TABLE default.columns_with_multiple_streams_compact\n(\n    `field0` Nullable(Int64) CODEC(Delta(2), LZ4),\n    `field1` Nullable(UInt8) CODEC(Delta(8), LZ4),\n    `field2` Array(Array(Int64)) CODEC(Delta(8), LZ4),\n    `field3` Tuple(\n        UInt32,\n        Array(UInt64)) CODEC(Delta, Default)\n)\nENGINE = MergeTree\nORDER BY tuple()\nSETTINGS min_rows_for_wide_part = 100000, min_bytes_for_wide_part = 100000, index_granularity = 8192
 1	1	[[1]]	(1,[1])
 2	2	[[2]]	(2,[2])
 3	3	[[3]]	(3,[3])
--- a/tests/queries/0_stateless/01548_create_table_compound_column_format.reference
+++ b/tests/queries/0_stateless/01548_create_table_compound_column_format.reference
@ -7,6 +7,7 @@ ENGINE = TinyLog
 CREATE TABLE test
 (
    `a` Int64,
-    `b` Tuple(a Int64)
+    `b` Tuple(
+        a Int64)
 )
 ENGINE = TinyLog
--- a/tests/queries/0_stateless/01603_read_with_backoff_bug.sql
+++ b/tests/queries/0_stateless/01603_read_with_backoff_bug.sql
@ -1,15 +1,17 @@
-- Tags: long, no-tsan, no-distributed-cache
-- Tag no-tsan: Too long for TSan
+-- Tags: long, no-tsan, no-msan, no-distributed-cache
+-- Too long for TSan and MSan

 set enable_filesystem_cache=0;
 set enable_filesystem_cache_on_write_operations=0;
+set max_rows_to_read = '30M';
+
 drop table if exists t;

 create table t (x UInt64, s String) engine = MergeTree order by x SETTINGS index_granularity = 8192, index_granularity_bytes = '10Mi';
 INSERT INTO t SELECT
    number,
    if(number < (8129 * 1024), arrayStringConcat(arrayMap(x -> toString(x), range(number % 128)), ' '), '')
-FROM numbers_mt((8129 * 1024) * 3) settings max_insert_threads=8;
+FROM numbers_mt((8129 * 1024) * 3) settings max_insert_threads=8, max_rows_to_read=0;

 -- optimize table t final;

--- a/tests/queries/0_stateless/01730_distributed_group_by_no_merge_order_by_long.sql
+++ b/tests/queries/0_stateless/01730_distributed_group_by_no_merge_order_by_long.sql
@ -1,6 +1,7 @@
 -- Tags: long, distributed, no-random-settings

 drop table if exists data_01730;
+SET max_rows_to_read = 0, max_result_rows = 0, max_bytes_before_external_group_by = 0;

 -- does not use 127.1 due to prefer_localhost_replica

--- a/tests/queries/0_stateless/01825_new_type_json_ghdata.sh
+++ b/tests/queries/0_stateless/01825_new_type_json_ghdata.sh
@ -1,5 +1,6 @@
 #!/usr/bin/env bash
-# Tags: no-fasttest
+# Tags: no-fasttest, no-s3-storage, long
+# ^ no-s3-storage: too memory hungry

 CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
@ -8,7 +9,8 @@ CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 ${CLICKHOUSE_CLIENT} -q "DROP TABLE IF EXISTS ghdata"
 ${CLICKHOUSE_CLIENT} -q "CREATE TABLE ghdata (data JSON) ENGINE = MergeTree ORDER BY tuple() SETTINGS index_granularity = 8192, index_granularity_bytes = '10Mi'" --allow_experimental_json_type 1

-cat $CUR_DIR/data_json/ghdata_sample.json | ${CLICKHOUSE_CLIENT} -q "INSERT INTO ghdata FORMAT JSONAsObject"
+cat $CUR_DIR/data_json/ghdata_sample.json | ${CLICKHOUSE_CLIENT} \
+  --max_memory_usage 10G --query "INSERT INTO ghdata FORMAT JSONAsObject"

 ${CLICKHOUSE_CLIENT} -q "SELECT count() FROM ghdata WHERE NOT ignore(*)"

@ -16,7 +18,7 @@ ${CLICKHOUSE_CLIENT} -q \
 "SELECT data.repo.name, count() AS stars FROM ghdata \
    WHERE data.type = 'WatchEvent' GROUP BY data.repo.name ORDER BY stars DESC, data.repo.name LIMIT 5"

-${CLICKHOUSE_CLIENT} --allow_experimental_analyzer=1 -q \
+${CLICKHOUSE_CLIENT} --enable_analyzer=1 -q \
 "SELECT data.payload.commits[].author.name AS name, count() AS c FROM ghdata \
    ARRAY JOIN data.payload.commits[].author.name \
    GROUP BY name ORDER BY c DESC, name LIMIT 5"
--- a/tests/queries/0_stateless/01825_new_type_json_ghdata_insert_select.sh
+++ b/tests/queries/0_stateless/01825_new_type_json_ghdata_insert_select.sh
@ -1,5 +1,6 @@
 #!/usr/bin/env bash
-# Tags: no-fasttest, long
+# Tags: no-fasttest, no-s3-storage, long
+# ^ no-s3-storage: it is memory-hungry

 CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
@ -13,10 +14,10 @@ ${CLICKHOUSE_CLIENT} -q "CREATE TABLE ghdata_2 (data JSON) ENGINE = MergeTree OR
 ${CLICKHOUSE_CLIENT} -q "CREATE TABLE ghdata_2_string (data String) ENGINE = MergeTree ORDER BY tuple() SETTINGS index_granularity = 8192, index_granularity_bytes = '10Mi'"
 ${CLICKHOUSE_CLIENT} -q "CREATE TABLE ghdata_2_from_string (data JSON) ENGINE = MergeTree ORDER BY tuple() SETTINGS index_granularity = 8192, index_granularity_bytes = '10Mi'" --allow_experimental_json_type 1

-cat $CUR_DIR/data_json/ghdata_sample.json | ${CLICKHOUSE_CLIENT} -q "INSERT INTO ghdata_2 FORMAT JSONAsObject"
+cat $CUR_DIR/data_json/ghdata_sample.json | ${CLICKHOUSE_CLIENT} --max_memory_usage 10G -q "INSERT INTO ghdata_2 FORMAT JSONAsObject"
 cat $CUR_DIR/data_json/ghdata_sample.json | ${CLICKHOUSE_CLIENT} -q "INSERT INTO ghdata_2_string FORMAT JSONAsString"

-${CLICKHOUSE_CLIENT} -q "INSERT INTO ghdata_2_from_string SELECT data FROM ghdata_2_string"
+${CLICKHOUSE_CLIENT} --max_memory_usage 10G -q "INSERT INTO ghdata_2_from_string SELECT data FROM ghdata_2_string"

 ${CLICKHOUSE_CLIENT} -q "SELECT \
    (SELECT mapSort(groupUniqArrayMap(JSONAllPathsWithTypes(data))), sum(cityHash64(toString(data))) FROM ghdata_2_from_string) = \
--- a/tests/queries/0_stateless/01825_type_json_ghdata.sh
+++ b/tests/queries/0_stateless/01825_type_json_ghdata.sh
@ -1,5 +1,6 @@
 #!/usr/bin/env bash
-# Tags: no-fasttest
+# Tags: no-fasttest, no-s3-storage, long
+# ^ no-s3-storage: it is memory-hungry

 CUR_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
 # shellcheck source=../shell_config.sh
--- a/Show More
+++ b/Show More