ClickHouse/programs/server/Server.cpp

#include <Server.h>
#include <Common/CurrentThread.h>
#include <Common/QueryScope.h>

#include <memory>
#include <Interpreters/ClientInfo.h>
#include <sys/resource.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <pwd.h>
#include <unistd.h>
#include <Poco/Net/HTTPServer.h>
#include <Poco/Net/NetException.h>
#include <Poco/Util/HelpFormatter.h>
#include <Poco/Util/LayeredConfiguration.h>
#include <Poco/Util/MapConfiguration.h>
#include <Poco/Util/SystemConfiguration.h>
#include <Poco/AutoPtr.h>
#include <Poco/Environment.h>
#include <Poco/Config.h>
#include <Common/AsynchronousMetricsKeyValuesMode.h>
#include <Common/ErrorCodes.h>
#include <Common/scope_guard_safe.h>
#include <Common/logger_useful.h>
#include <base/phdr_cache.h>
#include <Common/ErrorHandlers.h>
#include <Processors/QueryPlan/QueryPlanStepRegistry.h>
#include <base/getMemoryAmount.h>
#include <base/errnoToString.h>
#include <base/coverage.h>
#include <base/getFQDNOrHostName.h>
#include <base/safeExit.h>
#include <base/Numa.h>
#include <base/argsToConfig.h>
#include <Common/PoolId.h>
#include <Common/CurrentMemoryTracker.h>
#include <Common/MemoryPressureMonitor.h>
#include <Common/MemoryTracker.h>
#include <Common/PerCPUMemory.h>
#include <Common/MemoryWorker.h>
#include <Common/OOMCanary/OOMCanary.h>
#include <Common/ClickHouseRevision.h>
#include <Common/DNSResolver.h>
#include <Common/CgroupsMemoryUsageObserver.h>
#include <Common/CurrentMetrics.h>
#include <Common/DimensionalMetrics.h>
#include <Common/ISlotControl.h>
#include <Common/Macros.h>
#include <Common/ShellCommand.h>
#include <Common/ZooKeeper/ZooKeeper.h>
#include <Common/ZooKeeper/ZooKeeperNodeCache.h>
#include <Common/formatReadable.h>
#include <Common/getMultipleKeysFromConfig.h>
#include <Common/getNumberOfCPUCoresToUse.h>
#include <Common/getExecutablePath.h>
#include <Common/ProfileEvents.h>
#include <Common/Scheduler/IResourceManager.h>
#include <Common/ThreadProfileEvents.h>
#include <Common/ThreadStatus.h>
#include <Common/SilkFiberScheduler.h>
#include <Common/getMappedArea.h>
#include <Common/SignalHandlers.h>
#include <Common/remapExecutable.h>
#include <Common/SeccompFilter.h>
#include <Common/TLDListsHolder.h>
#include <Common/Config/AbstractConfigurationComparison.h>
#include <Common/Config/ConfigHelper.h>
#include <Common/assertProcessUserMatchesDataOwner.h>
#include <Common/makeSocketAddress.h>
#include <Common/FailPoint.h>
#include <Common/CPUID.h>
#include <Common/HTTPConnectionPool.h>
#include <Common/NamedCollections/NamedCollectionsFactory.h>
#include <Common/SQLDefinedHandlers/SQLDefinedHandlersFactory.h>
#include <Server/ConnectionRegistry.h>
#include <Server/createServer.h>
#include <Server/StartupWarnings.h>
#include <Server/socketBindListen.h>
#include <Server/stopServers.h>
#include <Server/waitServersToFinish.h>
#include <Interpreters/FileCache/FileCacheFactory.h>
#include <Core/BackgroundSchedulePool.h>
#include <Core/ServerSettings.h>
#include <Core/SettingsEnums.h>
#include <Core/ServerUUID.h>
#include <Core/Settings.h>
#include <IO/ReadBufferFromFile.h>
#include <IO/ReadBufferFromString.h>
#include <IO/ReadHelpers.h>
#include <IO/SharedThreadPools.h>
#include <IO/S3/Credentials.h>
#include <Interpreters/CancellationChecker.h>
#include <Interpreters/ServerAsynchronousMetrics.h>
#include <Interpreters/DDLWorker.h>
#include <Interpreters/DNSCacheUpdater.h>
#include <Interpreters/DatabaseCatalog.h>
#include <Interpreters/ExternalDictionariesLoader.h>
#include <Interpreters/ProcessList.h>
#include <Interpreters/executeQuery.h>
#include <Interpreters/loadMetadata.h>
#include <Interpreters/registerInterpreters.h>
#include <Interpreters/JIT/CompiledExpressionCache.h>
#include <Access/AccessControl.h>
#include <Access/ContextAccess.h>
#include <Access/User.h>
#include <Storages/MaterializedView/RefreshSet.h>
#include <Storages/MergeTree/MergeTreeBackgroundExecutor.h>
#include <Storages/MergeTree/MergeTreeSettings.h>
#include <Storages/MergeTree/ColumnsCache.h>
#include <Common/IMemoryReleasableCache.h>
#include <Storages/System/attachSystemTables.h>
#include <Storages/System/attachInformationSchemaTables.h>
#include <Storages/Cache/registerRemoteFileMetadatas.h>
#include <AggregateFunctions/registerAggregateFunctions.h>
#include <Functions/UserDefined/IUserDefinedSQLObjectsStorage.h>
#include <Functions/UserDefined/UserDefinedSQLFunctionFactory.h>
#include <Functions/pointInPolygon.h>
#include <Functions/registerFunctions.h>
#include <Parsers/registerParsers.h>
#include <TableFunctions/registerTableFunctions.h>
#include <Formats/registerFormats.h>
#include <Storages/registerStorages.h>
#include <Databases/registerDatabases.h>
#include <Dictionaries/registerDictionaries.h>
#include <Interpreters/SecretArgumentsRegistry.h>
#include <Disks/registerDisks.h>
#include <Common/Scheduler/Workload/IWorkloadEntityStorage.h>
#include <Coordination/KeeperContext.h>
#include <Common/Config/ConfigReloader.h>
#include <Server/HTTPHandlerFactory.h>
#include <Common/ReplicasReconnector.h>
#include <MetricsTransmitter.h>
#include <Common/StatusFile.h>
#include <Server/TCPHandlerFactory.h>
#include <Server/TCPServer.h>
#include <Common/SensitiveDataMasker.h>
#include <Common/ThreadFuzzer.h>
#include <Common/ThreadStackSize.h>
#include <Common/getHashOfLoadedBinary.h>
#include <Common/filesystemHelpers.h>
#include <Compression/CompressionCodecEncrypted.h>
#include <Parsers/ASTAlterQuery.h>
#include <Server/CloudPlacementInfo.h>
#include <Server/DistributedQuery/ExchangeConnections.h>
#include <Server/DistributedQuery/ExchangeServer.h>
#include <Server/HTTP/HTTPServer.h>
#include <Server/HTTP/HTTPServerConnectionFactory.h>
#include <Server/StatelessWorker/StatelessWorkerEndpoint.h>
#include <Server/MySQLHandlerFactory.h>
#include <Server/PostgreSQLHandlerFactory.h>
#include <Server/ProtocolServerAdapter.h>
#include <Server/PrometheusRequestHandlerFactory.h>
#include <Server/ProxyV1HandlerFactory.h>
#include <Server/TLSHandlerFactory.h>
#include <Server/KeeperHTTPHandlerFactory.h>
#include <Server/IcebergRESTCatalog/IcebergRESTCatalogHandlerFactory.h>
#include <Server/ArrowFlight/ArrowFlightServer.h>
#include <Interpreters/AsynchronousInsertQueue.h>

#include <filesystem>
#include <unordered_set>

#include <Common/Jemalloc.h>
#include <Common/JemallocCacheArena.h>
#include <Common/JemallocMergeTreeArena.h>

#include "config.h"
#include <Common/config_version.h>

#if defined(OS_LINUX)
#    include <cstdlib>
#    include <sys/un.h>
#    include <sys/mman.h>
#    include <Common/hasLinuxCapability.h>
#endif

#if USE_SSL
#    include <Poco/Net/SecureServerSocket.h>
#    include <Server/CertificateReloader.h>
#    include <Server/ACME/Client.h>
#endif

#if USE_SSH && defined(OS_LINUX)
#    include <Server/SSH/SSHPtyHandlerFactory.h>
#    include <Common/LibSSHInitializer.h>
#    include <Common/LibSSHLogger.h>
#endif

#if USE_GRPC
#   include <Server/GRPCServer.h>
#endif

#if USE_NURAFT
#    include <Coordination/FourLetterCommand.h>
#    include <Coordination/KeeperAsynchronousMetrics.h>
#    include <Server/KeeperTCPHandlerFactory.h>
#endif

#if USE_AZURE_BLOB_STORAGE
#   include <azure/storage/common/internal/xml_wrapper.hpp>
#   include <azure/core/diagnostics/logger.hpp>
#endif


/// A minimal file used when the server is run without installation
constexpr unsigned char resource_embedded_xml[] =
{
#embed "embedded.xml"
};

namespace DB
{
namespace Setting
{
    extern const SettingsSeconds http_receive_timeout;
    extern const SettingsSeconds http_send_timeout;
    extern const SettingsSeconds receive_timeout;
    extern const SettingsSeconds send_timeout;
}

namespace ServerSetting
{
    extern const ServerSettingsUInt32 allow_feature_tier;
    extern const ServerSettingsUInt32 asynchronous_heavy_metrics_update_period_s;
    extern const ServerSettingsUInt32 asynchronous_metrics_update_period_s;
    extern const ServerSettingsBool asynchronous_metrics_enable_heavy_metrics;
    extern const ServerSettingsBool asynchronous_metrics_keeper_metrics_only;
    extern const ServerSettingsBool async_insert_queue_flush_on_shutdown;
    extern const ServerSettingsUInt64 async_insert_threads;
    extern const ServerSettingsBool async_load_databases;
    extern const ServerSettingsBool async_load_system_database;
    extern const ServerSettingsUInt64 background_buffer_flush_schedule_pool_size;
    extern const ServerSettingsUInt64 background_common_pool_size;
    extern const ServerSettingsUInt64 background_distributed_schedule_pool_size;
    extern const ServerSettingsUInt64 background_fetches_pool_size;
    extern const ServerSettingsFloat background_merges_mutations_concurrency_ratio;
    extern const ServerSettingsString background_merges_mutations_scheduling_policy;
    extern const ServerSettingsUInt64 background_message_broker_schedule_pool_size;
    extern const ServerSettingsUInt64 background_move_pool_size;
    extern const ServerSettingsUInt64 background_pool_size;
    extern const ServerSettingsUInt64 background_schedule_pool_size;
    extern const ServerSettingsUInt64 background_streaming_schedule_pool_size;
    extern const ServerSettingsUInt64 backups_io_thread_pool_queue_size;
    extern const ServerSettingsDouble cache_size_to_ram_max_ratio;
    extern const ServerSettingsString columns_cache_policy;
    extern const ServerSettingsUInt64 columns_cache_size;
    extern const ServerSettingsDouble columns_cache_size_ratio;
    extern const ServerSettingsDouble columns_cache_size_to_ram_ratio;
    extern const ServerSettingsDouble columns_cache_free_memory_ratio;
    extern const ServerSettingsUInt64 columns_cache_history_window_ms;
    extern const ServerSettingsDouble cannot_allocate_thread_fault_injection_probability;
    extern const ServerSettingsUInt64 cgroups_memory_usage_observer_wait_time;
    extern const ServerSettingsUInt64 compiled_expression_cache_elements_size;
    extern const ServerSettingsUInt64 compiled_expression_cache_size;
    extern const ServerSettingsUInt64 concurrent_threads_soft_limit_num;
    extern const ServerSettingsUInt64 concurrent_threads_soft_limit_ratio_to_cores;
    extern const ServerSettingsString concurrent_threads_scheduler;
    extern const ServerSettingsBool concurrent_threads_lazy_allocation;
    extern const ServerSettingsUInt64 config_reload_interval_ms;
    extern const ServerSettingsUInt64 database_catalog_drop_table_concurrency;
    extern const ServerSettingsUInt64 database_catalog_shutdown_table_concurrency;
    extern const ServerSettingsString default_database;
    extern const ServerSettingsString insert_deduplication_version;
    extern const ServerSettingsBool disable_internal_dns_cache;
    extern const ServerSettingsBool s3queue_disable_streaming;
    extern const ServerSettingsBool enable_read_through_distributed_cache;
    extern const ServerSettingsBool enable_write_through_distributed_cache;
    extern const ServerSettingsUInt64 disk_connections_soft_limit;
    extern const ServerSettingsUInt64 disk_connections_store_limit;
    extern const ServerSettingsUInt64 disk_connections_hard_limit;
    extern const ServerSettingsUInt64 disk_connections_warn_limit;
    extern const ServerSettingsUInt64 disk_connections_rcvbuf;
    extern const ServerSettingsUInt64 disk_connections_sndbuf;
    extern const ServerSettingsBool dns_allow_resolve_names_to_ipv4;
    extern const ServerSettingsBool dns_allow_resolve_names_to_ipv6;
    extern const ServerSettingsUInt64 dns_cache_max_entries;
    extern const ServerSettingsInt32 dns_cache_update_period;
    extern const ServerSettingsUInt32 dns_max_consecutive_failures;
    extern const ServerSettingsBool enable_azure_sdk_logging;
    extern const ServerSettingsBool enable_silk_runtime;
    extern const ServerSettingsBool enable_gpu;
    extern const ServerSettingsUInt64 global_profiler_cpu_time_period_ns;
    extern const ServerSettingsUInt64 global_profiler_real_time_period_ns;
    extern const ServerSettingsUInt64 http_connections_soft_limit;
    extern const ServerSettingsUInt64 http_connections_store_limit;
    extern const ServerSettingsUInt64 http_connections_hard_limit;
    extern const ServerSettingsUInt64 http_connections_warn_limit;
    extern const ServerSettingsUInt64 http_connections_rcvbuf;
    extern const ServerSettingsUInt64 http_connections_sndbuf;
    extern const ServerSettingsString index_mark_cache_policy;
    extern const ServerSettingsUInt64 index_mark_cache_size;
    extern const ServerSettingsDouble index_mark_cache_size_ratio;
    extern const ServerSettingsString vector_similarity_index_cache_policy;
    extern const ServerSettingsUInt64 vector_similarity_index_cache_size;
    extern const ServerSettingsUInt64 vector_similarity_index_cache_max_entries;
    extern const ServerSettingsDouble vector_similarity_index_cache_size_ratio;
    extern const ServerSettingsString text_index_tokens_cache_policy;
    extern const ServerSettingsUInt64 text_index_tokens_cache_size;
    extern const ServerSettingsUInt64 text_index_tokens_cache_max_entries;
    extern const ServerSettingsDouble text_index_tokens_cache_size_ratio;
    extern const ServerSettingsString text_index_header_cache_policy;
    extern const ServerSettingsUInt64 text_index_header_cache_size;
    extern const ServerSettingsUInt64 text_index_header_cache_max_entries;
    extern const ServerSettingsDouble text_index_header_cache_size_ratio;
    extern const ServerSettingsString text_index_postings_cache_policy;
    extern const ServerSettingsUInt64 text_index_postings_cache_size;
    extern const ServerSettingsUInt64 text_index_postings_cache_max_entries;
    extern const ServerSettingsDouble text_index_postings_cache_size_ratio;
    extern const ServerSettingsString index_uncompressed_cache_policy;
    extern const ServerSettingsUInt64 index_uncompressed_cache_size;
    extern const ServerSettingsDouble index_uncompressed_cache_size_ratio;
    extern const ServerSettingsString iceberg_metadata_files_cache_policy;
    extern const ServerSettingsUInt64 iceberg_metadata_files_cache_size;
    extern const ServerSettingsUInt64 iceberg_metadata_files_cache_max_entries;
    extern const ServerSettingsDouble iceberg_metadata_files_cache_size_ratio;
    extern const ServerSettingsString paimon_metadata_files_cache_policy;
    extern const ServerSettingsUInt64 paimon_metadata_files_cache_size;
    extern const ServerSettingsUInt64 paimon_metadata_files_cache_max_entries;
    extern const ServerSettingsDouble paimon_metadata_files_cache_size_ratio;
    extern const ServerSettingsString parquet_metadata_cache_policy;
    extern const ServerSettingsUInt64 parquet_metadata_cache_size;
    extern const ServerSettingsUInt64 parquet_metadata_cache_max_entries;
    extern const ServerSettingsDouble parquet_metadata_cache_size_ratio;
    extern const ServerSettingsUInt64 io_thread_pool_queue_size;
    extern const ServerSettingsBool jemalloc_enable_global_profiler;
    extern const ServerSettingsBool jemalloc_collect_global_profile_samples_in_trace_log;
    extern const ServerSettingsBool jemalloc_enable_background_threads;
    extern const ServerSettingsUInt64 jemalloc_max_background_threads_num;
    extern const ServerSettingsUInt64 jemalloc_profiler_sampling_rate;
    extern const ServerSettingsSeconds keep_alive_timeout;
    extern const ServerSettingsString mark_cache_policy;
    extern const ServerSettingsUInt64 mark_cache_size;
    extern const ServerSettingsDouble mark_cache_size_ratio;
    extern const ServerSettingsString unique_key_index_cache_policy;
    extern const ServerSettingsUInt64 unique_key_index_cache_size_bytes;
    extern const ServerSettingsDouble unique_key_index_cache_size_ratio;
    extern const ServerSettingsString unique_key_bitmap_cache_policy;
    extern const ServerSettingsUInt64 unique_key_bitmap_cache_size_bytes;
    extern const ServerSettingsDouble unique_key_bitmap_cache_size_ratio;
    extern const ServerSettingsUInt64 max_fetch_partition_thread_pool_size;
    extern const ServerSettingsUInt64 max_active_parts_loading_thread_pool_size;
    extern const ServerSettingsUInt64 max_backups_io_thread_pool_free_size;
    extern const ServerSettingsUInt64 max_backups_io_thread_pool_size;
    extern const ServerSettingsUInt64 max_concurrent_insert_queries;
    extern const ServerSettingsUInt64 max_concurrent_queries;
    extern const ServerSettingsUInt64 max_concurrent_select_queries;
    extern const ServerSettingsInt32 max_connections;
    extern const ServerSettingsUInt64 max_database_num_to_warn;
    extern const ServerSettingsUInt32 max_database_replicated_create_table_thread_pool_size;
    extern const ServerSettingsUInt64 max_dictionary_num_to_warn;
    extern const ServerSettingsUInt64 max_io_thread_pool_free_size;
    extern const ServerSettingsUInt64 max_io_thread_pool_size;
    extern const ServerSettingsUInt64 max_keep_alive_requests;
    extern const ServerSettingsUInt64 max_outdated_parts_loading_thread_pool_size;
    extern const ServerSettingsUInt64 max_per_cpu_untracked_memory;
    extern const ServerSettingsUInt64 max_partition_size_to_drop;
    extern const ServerSettingsUInt64 max_part_num_to_warn;
    extern const ServerSettingsUInt64 max_pending_mutations_to_warn;
    extern const ServerSettingsUInt64 max_pending_mutations_execution_time_to_warn;
    extern const ServerSettingsUInt64 max_parts_cleaning_thread_pool_size;
    extern const ServerSettingsUInt64 max_named_collection_num_to_warn;
    extern const ServerSettingsUInt64 max_named_collection_num_to_throw;
    extern const ServerSettingsUInt64 max_table_num_to_throw;
    extern const ServerSettingsUInt64 max_replicated_table_num_to_throw;
    extern const ServerSettingsUInt64 max_view_num_to_throw;
    extern const ServerSettingsUInt64 max_dictionary_num_to_throw;
    extern const ServerSettingsUInt64 max_database_num_to_throw;
    extern const ServerSettingsUInt64 max_remote_read_network_bandwidth_for_server;
    extern const ServerSettingsUInt64 max_remote_write_network_bandwidth_for_server;
    extern const ServerSettingsUInt64 max_remote_read_connections;
    extern const ServerSettingsUInt64 reader_executor_memory_pressure_elevated_level_pct;
    extern const ServerSettingsUInt64 reader_executor_memory_pressure_high_level_pct;
    extern const ServerSettingsUInt64 reader_executor_memory_pressure_critical_level_pct;
    extern const ServerSettingsUInt64 max_local_read_bandwidth_for_server;
    extern const ServerSettingsUInt64 max_local_write_bandwidth_for_server;
    extern const ServerSettingsUInt64 max_server_memory_usage;
    extern const ServerSettingsDouble max_server_memory_usage_to_ram_ratio;
    extern const ServerSettingsUInt64 max_table_num_to_warn;
    extern const ServerSettingsUInt64 max_table_size_to_drop;
    extern const ServerSettingsUInt64 max_temporary_data_on_disk_size;
    extern const ServerSettingsUInt64 max_thread_pool_free_size;
    extern const ServerSettingsUInt64 max_thread_pool_size;
    extern const ServerSettingsUInt64 max_unexpected_parts_loading_thread_pool_size;
    extern const ServerSettingsUInt64 max_view_num_to_warn;
    extern const ServerSettingsUInt64 max_waiting_queries;
    extern const ServerSettingsUInt64 memory_worker_period_ms;
    extern const ServerSettingsDouble memory_worker_purge_dirty_pages_threshold_ratio;
    extern const ServerSettingsDouble memory_worker_purge_total_memory_threshold_ratio;
    extern const ServerSettingsUInt64 memory_worker_decay_adjustment_period_ms;
    extern const ServerSettingsBool memory_worker_correct_memory_tracker;
    extern const ServerSettingsBool memory_worker_use_cgroup;
    extern const ServerSettingsDouble memory_worker_rss_speculative_reserve_ratio;
    extern const ServerSettingsBool memory_worker_dynamic_hard_limit;
    extern const ServerSettingsUInt64 merges_mutations_memory_usage_soft_limit;
    extern const ServerSettingsDouble merges_mutations_memory_usage_to_ram_ratio;
    extern const ServerSettingsString merge_workload;
    extern const ServerSettingsUInt64 min_allocation_size_to_log_stack_trace;
    extern const ServerSettingsUInt64 min_allocation_size_to_throw_on_memory_limit;
    extern const ServerSettingsUInt64 mmap_cache_size;
    extern const ServerSettingsString mutation_workload;
    extern const ServerSettingsString query_condition_cache_policy;
    extern const ServerSettingsUInt64 query_condition_cache_size;
    extern const ServerSettingsDouble query_condition_cache_size_ratio;
    extern const ServerSettingsString encryption_header_cache_policy;
    extern const ServerSettingsUInt64 encryption_header_cache_size;
    extern const ServerSettingsDouble encryption_header_cache_size_ratio;
    extern const ServerSettingsBool prepare_system_log_tables_on_startup;
    extern const ServerSettingsBool user_profile_events_per_cpu;
    extern const ServerSettingsBool show_addresses_in_stack_traces;
    extern const ServerSettingsBool shutdown_wait_backups_and_restores;
    extern const ServerSettingsUInt64 shutdown_wait_unfinished;
    extern const ServerSettingsBool shutdown_wait_unfinished_queries;
    extern const ServerSettingsUInt64 storage_connections_soft_limit;
    extern const ServerSettingsUInt64 storage_connections_store_limit;
    extern const ServerSettingsUInt64 storage_connections_hard_limit;
    extern const ServerSettingsUInt64 storage_connections_warn_limit;
    extern const ServerSettingsUInt64 storage_connections_rcvbuf;
    extern const ServerSettingsUInt64 storage_connections_sndbuf;
    extern const ServerSettingsUInt64 tables_loader_background_pool_size;
    extern const ServerSettingsUInt64 tables_loader_foreground_pool_size;
    extern const ServerSettingsString temporary_data_in_cache;
    extern const ServerSettingsUInt64 thread_pool_queue_size;
    extern const ServerSettingsString tmp_policy;
    extern const ServerSettingsUInt64 total_memory_profiler_sample_max_allocation_size;
    extern const ServerSettingsUInt64 total_memory_profiler_sample_min_allocation_size;
    extern const ServerSettingsUInt64 total_memory_profiler_step;
    extern const ServerSettingsDouble total_memory_tracker_sample_probability;
    extern const ServerSettingsBool throw_on_unknown_workload;
    extern const ServerSettingsBool cpu_slot_preemption;
    extern const ServerSettingsUInt64 cpu_slot_quantum_ns;
    extern const ServerSettingsUInt64 cpu_slot_preemption_timeout_ms;
    extern const ServerSettingsBool cpu_slot_parking;
    extern const ServerSettingsString uncompressed_cache_policy;
    extern const ServerSettingsUInt64 uncompressed_cache_size;
    extern const ServerSettingsDouble uncompressed_cache_size_ratio;
    extern const ServerSettingsUInt64 per_cpu_untracked_memory_thread_buffer;
    extern const ServerSettingsBool use_separate_cache_arena;
    extern const ServerSettingsString primary_index_cache_policy;
    extern const ServerSettingsUInt64 primary_index_cache_size;
    extern const ServerSettingsDouble primary_index_cache_size_ratio;
    extern const ServerSettingsString statistics_cache_policy;
    extern const ServerSettingsUInt64 statistics_cache_size;
    extern const ServerSettingsDouble statistics_cache_size_ratio;
    extern const ServerSettingsUInt64 point_in_polygon_cache_size;
    extern const ServerSettingsBool dictionaries_lazy_load;
    extern const ServerSettingsBool wait_dictionaries_load_at_startup;
    extern const ServerSettingsUInt64 max_prefixes_deserialization_thread_pool_size;
    extern const ServerSettingsUInt64 max_prefixes_deserialization_thread_pool_free_size;
    extern const ServerSettingsUInt64 prefixes_deserialization_thread_pool_thread_pool_queue_size;
    extern const ServerSettingsUInt64 max_format_parsing_thread_pool_size;
    extern const ServerSettingsUInt64 max_format_parsing_thread_pool_free_size;
    extern const ServerSettingsUInt64 format_parsing_thread_pool_queue_size;
    extern const ServerSettingsNonZeroUInt64 ai_function_max_thread_pool_size;
    extern const ServerSettingsUInt64 ai_function_max_thread_pool_free_size;
    extern const ServerSettingsUInt64 ai_function_thread_pool_queue_size;
    extern const ServerSettingsUInt64 page_cache_history_window_ms;
    extern const ServerSettingsString page_cache_policy;
    extern const ServerSettingsDouble page_cache_size_ratio;
    extern const ServerSettingsUInt64 page_cache_min_size;
    extern const ServerSettingsUInt64 page_cache_max_size;
    extern const ServerSettingsDouble page_cache_free_memory_ratio;
    extern const ServerSettingsUInt64 page_cache_shards;
    extern const ServerSettingsUInt64 os_cpu_busy_time_threshold;
    extern const ServerSettingsFloat min_os_cpu_wait_time_ratio_to_drop_connection;
    extern const ServerSettingsFloat max_os_cpu_wait_time_ratio_to_drop_connection;
    extern const ServerSettingsBool skip_binary_checksum_checks;
    extern const ServerSettingsBool abort_on_logical_error;
    extern const ServerSettingsBool collect_connection_metrics;
    extern const ServerSettingsUInt64 jemalloc_flush_profile_interval_bytes;
    extern const ServerSettingsUInt64 jemalloc_merge_tree_arenas;
    extern const ServerSettingsBool jemalloc_flush_profile_on_memory_exceeded;
    extern const ServerSettingsUInt64 jemalloc_flush_profile_on_memory_exceeded_interval;
    extern const ServerSettingsString allowed_disks_for_table_engines;
    extern const ServerSettingsUInt64 s3_credentials_provider_max_cache_size;
    extern const ServerSettingsUInt64 max_open_files;
    extern const ServerSettingsString path;
    extern const ServerSettingsString user_files_path;
    extern const ServerSettingsString user_scripts_path;
    extern const ServerSettingsString dynamic_user_defined_executable_functions_path;
    extern const ServerSettingsString top_level_domains_path;
    extern const ServerSettingsString interserver_http_host;
    extern const ServerSettingsUInt64 interserver_http_port;
    extern const ServerSettingsString interserver_https_host;
    extern const ServerSettingsUInt64 interserver_https_port;
    extern const ServerSettingsString include_from;
    extern const ServerSettingsString tmp_path;
    extern const ServerSettingsString format_schema_path;
    extern const ServerSettingsString google_protos_path;
    extern const ServerSettingsString filesystem_caches_path;
    extern const ServerSettingsInt32 oom_score;
    extern const ServerSettingsBool oom_canary_enable;
    extern const ServerSettingsUInt64 oom_canary_size;
    extern const ServerSettingsBool oom_canary_relaunch;
    extern const ServerSettingsUInt64 oom_canary_max_rapid_relaunches;
    extern const ServerSettingsUInt64 oom_canary_initial_backoff_seconds;
    extern const ServerSettingsUInt64 oom_canary_max_backoff_seconds;
    extern const ServerSettingsBool remap_executable;
    extern const ServerSettingsBool mlock_executable;
    extern const ServerSettingsUInt64 mlock_executable_min_total_memory_amount_bytes;
    extern const ServerSettingsSeccompMode seccomp;
    extern const ServerSettingsUInt32 listen_backlog;
    extern const ServerSettingsBool listen_reuse_port;
    extern const ServerSettingsBool listen_try;
    extern const ServerSettingsBool mysql_require_secure_transport;
    extern const ServerSettingsBool postgresql_require_secure_transport;
    extern const ServerSettingsBool skip_check_for_incorrect_settings;
    extern const ServerSettingsUInt64 query_cache_max_size_in_bytes;
    extern const ServerSettingsUInt64 query_cache_max_entries;
    extern const ServerSettingsUInt64 query_cache_max_entry_size_in_bytes;
    extern const ServerSettingsUInt64 query_cache_max_entry_size_in_rows;
    extern const ServerSettingsString logger_log;
    extern const ServerSettingsString logger_level;
    extern const ServerSettingsString logger_startup_level;
    extern const ServerSettingsString logger_shutdown_level;
    extern const ServerSettingsString openssl_server_certificate_file;
    extern const ServerSettingsString openssl_server_private_key_file;
    extern const ServerSettingsString openssl_server_ca_config;
    extern const ServerSettingsString openssl_client_certificate_file;
    extern const ServerSettingsString openssl_client_private_key_file;
    extern const ServerSettingsString openssl_client_ca_config;
    extern const ServerSettingsString distributed_ddl_path;
    extern const ServerSettingsString distributed_ddl_replicas_path;
    extern const ServerSettingsInt32 distributed_ddl_pool_size;
    extern const ServerSettingsBool prometheus_keeper_metrics_only;
    extern const ServerSettingsBool startup_scripts_throw_on_error;
    extern const ServerSettingsUInt64 keeper_server_socket_receive_timeout_sec;
    extern const ServerSettingsUInt64 keeper_server_socket_send_timeout_sec;
    extern const ServerSettingsString hdfs_libhdfs3_conf;
    extern const ServerSettingsString config_file;
    extern const ServerSettingsString users_to_ignore_early_memory_limit_check;
}

namespace ErrorCodes
{
    extern const int STARTUP_SCRIPTS_ERROR;
}

namespace FileCacheSetting
{
    extern const FileCacheSettingsBool load_metadata_asynchronously;
}

}

namespace CurrentMetrics
{
    extern const Metric Revision;
    extern const Metric VersionInteger;
    extern const Metric MemoryTracking;
    extern const Metric MergesMutationsMemoryTracking;
    extern const Metric MaxDDLEntryID;
    extern const Metric MaxPushedDDLEntryID;
    extern const Metric StartupScriptsExecutionState;
    extern const Metric IsServerShuttingDown;
}

namespace DimensionalMetrics
{
    extern MetricFamily & StartupScriptsFailureReason;
}

namespace ProfileEvents
{
    extern const Event MainConfigLoads;
    extern const Event ServerStartupMilliseconds;
    extern const Event InterfaceNativeSendBytes;
    extern const Event InterfaceNativeReceiveBytes;
    extern const Event InterfaceHTTPSendBytes;
    extern const Event InterfaceHTTPReceiveBytes;
    extern const Event InterfacePrometheusSendBytes;
    extern const Event InterfacePrometheusReceiveBytes;
    extern const Event InterfaceInterserverSendBytes;
    extern const Event InterfaceInterserverReceiveBytes;
    extern const Event InterfaceMySQLSendBytes;
    extern const Event InterfaceMySQLReceiveBytes;
    extern const Event InterfacePostgreSQLSendBytes;
    extern const Event InterfacePostgreSQLReceiveBytes;
}

namespace fs = std::filesystem;

int mainEntryClickHouseServer(int argc, char ** argv);
int mainEntryClickHouseServer(int argc, char ** argv)
{
    DB::Server app;

    /// Do not fork separate process from watchdog if we attached to terminal.
    /// Otherwise it breaks gdb usage.
    /// Can be overridden by environment variable (cannot use server config at this moment).
    if (argc > 0)
    {
        const char * env_watchdog = getenv("CLICKHOUSE_WATCHDOG_ENABLE"); // NOLINT(concurrency-mt-unsafe)
        if (env_watchdog)
        {
            if (0 == strcmp(env_watchdog, "1"))
                app.shouldSetupWatchdog(argv[0]);

            /// Other values disable watchdog explicitly.
        }
        else if (!isatty(STDIN_FILENO) && !isatty(STDOUT_FILENO) && !isatty(STDERR_FILENO))
            app.shouldSetupWatchdog(argv[0]);
    }

    try
    {
        return app.run(argc, argv);
    }
    catch (...)
    {
        std::cerr << DB::getCurrentExceptionMessage(true) << "\n";
        auto code = DB::getCurrentExceptionCode();
        return static_cast<UInt8>(code) ? code : 1;
    }
}

namespace DB
{

namespace ErrorCodes
{
    extern const int ABORTED;
    extern const int NO_ELEMENTS_IN_CONFIG;
    extern const int SUPPORT_IS_DISABLED;
    extern const int ARGUMENT_OUT_OF_BOUND;
    extern const int EXCESSIVE_ELEMENT_IN_CONFIG;
    extern const int INVALID_CONFIG_PARAMETER;
    extern const int INVALID_SETTING_VALUE;
    extern const int NETWORK_ERROR;
    extern const int CORRUPTED_DATA;
    extern const int BAD_ARGUMENTS;
}


enum StartupScriptsExecutionState : CurrentMetrics::Value
{
    NotFinished = 0,
    Success = 1,
    Failure = 2,
};


namespace
{

std::string getCanonicalPath(std::string && path, const std::string & base = {})
{
    Poco::trimInPlace(path);
    if (path.empty())
        throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "path configuration parameter is empty");
    if (!base.empty() && fs::path(path).is_relative())
        path = fs::weakly_canonical(fs::path(base) / path);
    if (path.back() != '/')
        path += '/';
    return std::move(path);
}

Poco::Net::TCPServerParams::Ptr makeServerParams(const ServerSettings & server_settings)
{
    Poco::Net::TCPServerParams::Ptr params = new Poco::Net::TCPServerParams();
    params->setMaxQueued(server_settings[ServerSetting::listen_backlog]);
    return params;
}

#if defined(OS_LINUX)
/// Whether a debugger is attached to this process, according to the `TracerPid` field of
/// `/proc/self/status`, which the kernel sets to the pid of the tracer and to zero when there is
/// none. This is a plain read, so unlike the `ptrace(PTRACE_TRACEME)` probe it used to be, it
/// answers the same way whatever the `seccomp` server setting denies.
bool isRunUnderDebugger()
{
    try
    {
        ReadBufferFromFile status("/proc/self/status");
        while (!status.eof())
        {
            String line;
            readStringUntilNewlineInto(line, status);
            if (!status.eof())
                ++status.position();

            static constexpr std::string_view prefix = "TracerPid:";
            if (!line.starts_with(prefix))
                continue;

            UInt64 tracer_pid = 0;
            ReadBufferFromString value(line);
            value.ignore(prefix.size());
            skipWhitespaceIfAny(value);
            readIntText(tracer_pid, value);
            return tracer_pid != 0;
        }
    }
    catch (...)
    {
        tryLogCurrentException(__PRETTY_FUNCTION__);
    }

    return false;
}
#endif

}

Poco::Net::SocketAddress Server::socketBindListen(
    const ServerSettings & server_settings,
    Poco::Net::ServerSocket & socket,
    const std::string & host,
    UInt16 port,
    [[maybe_unused]] bool secure) const
{
    return DB::socketBindListen(server_settings, socket, host, port, &logger());
}

namespace
{

Strings getListenHosts(const Poco::Util::AbstractConfiguration & config)
{
    auto listen_hosts = DB::getMultipleValuesFromConfig(config, "", "listen_host");
    if (listen_hosts.empty())
    {
        listen_hosts.emplace_back("::1");
        listen_hosts.emplace_back("127.0.0.1");
    }
    return listen_hosts;
}

Strings getInterserverListenHosts(const Poco::Util::AbstractConfiguration & config)
{
    auto interserver_listen_hosts = DB::getMultipleValuesFromConfig(config, "", "interserver_listen_host");
    if (!interserver_listen_hosts.empty())
      return interserver_listen_hosts;

    /// Use more general restriction in case of emptiness
    return getListenHosts(config);
}

bool isIntrospectionProtocol(const Poco::Util::AbstractConfiguration & config, const std::string & protocol)
{
    return config.getBool("protocols." + protocol + ".introspection", false);
}

bool hasIntrospectionProtocols(const Poco::Util::AbstractConfiguration & config)
{
    Poco::Util::AbstractConfiguration::Keys protocols;
    config.keys("protocols", protocols);
    return std::any_of(protocols.begin(), protocols.end(), [&](const auto & protocol)
    {
        return isIntrospectionProtocol(config, protocol);
    });
}

bool getListenTry(const Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings)
{
    bool listen_try = server_settings[ServerSetting::listen_try];
    if (!listen_try)
    {
        Poco::Util::AbstractConfiguration::Keys protocols;
        config.keys("protocols", protocols);
        listen_try =
            DB::getMultipleValuesFromConfig(config, "", "listen_host").empty() &&
            std::none_of(protocols.begin(), protocols.end(), [&](const auto & protocol)
            {
                return config.has("protocols." + protocol + ".host") && config.has("protocols." + protocol + ".port");
            });
    }
    return listen_try;
}

#if USE_GRPC || USE_ARROWFLIGHT
bool isWildcardListenHost(const std::string & listen_host)
{
    Poco::Net::IPAddress address;
    return Poco::Net::IPAddress::tryParse(listen_host, address) && address.isWildcard();
}

/// gRPC treats `::` and `0.0.0.0` as one family-agnostic wildcard: for either of them it binds a
/// dual-stack `::` socket and falls back to `0.0.0.0` only on a host without IPv6 (see
/// `add_wildcard_addrs_to_server` in `contrib/grpc`). A wildcard listener therefore overlaps every
/// listener on that port, not only another wildcard listener. Normalize the gRPC-based listeners to
/// the first wildcard address so a configuration such as `127.0.0.1`, `0.0.0.0`, `::1` creates one
/// listener instead of creating overlapping sockets. This applies independently to each gRPC-based
/// port because all such ports use the same `listen_host` configuration.
Strings getGRPCListenHosts(const Strings & listen_hosts)
{
    const auto wildcard_it = std::ranges::find_if(listen_hosts, isWildcardListenHost);
    if (wildcard_it == listen_hosts.end())
        return listen_hosts;

    return {*wildcard_it};
}
#endif

}


void Server::createServer(
    Poco::Util::AbstractConfiguration & config,
    const std::string & listen_host,
    const char * port_name,
    bool listen_try,
    bool start_server,
    std::vector<ProtocolServerAdapter> & servers,
    CreateServerFunc && func) const
{
    if (DB::createServer(config, listen_host, port_name, listen_try, start_server, servers, std::move(func), &logger())
        && (start_server || !servers.back().bindsOnStart()))
    {
        /// Register the configured port rather than the actual bound port. `getServerPort` keeps a
        /// single value per `port_name`, so with `tcp_port=0` (OS-assigned) and several `listen_host`
        /// values (e.g. the default `::1` + `127.0.0.1`) each host binds a distinct ephemeral port and
        /// registering the actual port would let the last host overwrite the others, leaving
        /// `getServerPort` pointing at a port that is not listening on the host a client uses. When the
        /// configured port is non-zero it equals the bound port anyway, so this preserves the previous
        /// behavior in all cases. (`clickhouse-local` registers the actual bound port because it needs
        /// the OS-assigned value, but it rejects the ambiguous `port=0` + multiple `listen_host` combo.)
        global_context->registerServerPort(port_name, static_cast<UInt16>(config.getInt(port_name)));
    }
}


#if defined(OS_LINUX)
namespace
{

void setOOMScore(int value, LoggerRawPtr log)
{
    try
    {
        std::string value_string = std::to_string(value);
        DB::WriteBufferFromFile buf("/proc/self/oom_score_adj");
        buf.write(value_string.c_str(), value_string.size());
        buf.next();
        buf.close();
    }
    catch (const Poco::Exception & e)
    {
        LOG_WARNING(log, "Failed to adjust OOM score: '{}'.", e.displayText());
        return;
    }
    LOG_INFO(log, "Set OOM score adjustment to {}", value);
}

}
#endif


void Server::uninitialize()
{
    logger().information("shutting down");
    BaseDaemon::uninitialize();
}

int Server::run()
{
    if (config().hasOption("help"))
    {
        Poco::Util::HelpFormatter help_formatter(Server::options());
        std::string app_name = (commandName() == "clickhouse-server") ? "clickhouse-server" : "clickhouse server";
        auto header_str = fmt::format("{} [OPTION] [-- [ARG]...]\n"
                                      "positional arguments can be used to rewrite config.xml properties, for example, --http_port=8010",
                                      app_name);
        help_formatter.setHeader(header_str);
        help_formatter.format(std::cout);
        return 0;
    }
    if (config().hasOption("version"))
    {
        std::cout << VERSION_NAME << " server version " << VERSION_STRING << VERSION_OFFICIAL << "." << std::endl;
        return 0;
    }
    return Application::run(); // NOLINT
}

void Server::initialize(Poco::Util::Application & self)
{
    Poco::AutoPtr<Poco::Util::MapConfiguration> properties(new Poco::Util::MapConfiguration);
    Poco::Util::AbstractConfiguration::Keys property_keys;
    config().keys("application", property_keys);
    for (const auto & key : property_keys)
        properties->setString("application." + key, config().getRawString("application." + key));
    application_properties = properties;

    ConfigProcessor::registerEmbeddedConfig("config.xml", std::string_view(reinterpret_cast<const char *>(resource_embedded_xml), std::size(resource_embedded_xml)));
    BaseDaemon::initialize(self);
    logger().information("starting up");

    LOG_INFO(&logger(), "OS name: {}, version: {}, architecture: {}",
        Poco::Environment::osName(),
        Poco::Environment::osVersion(),
        Poco::Environment::osArchitecture());
}

std::string Server::getDefaultCorePath() const
{
    return getCanonicalPath(config().getString("path", DBMS_DEFAULT_PATH), original_working_directory) + "cores";
}

void Server::defineOptions(Poco::Util::OptionSet & options)
{
    options.addOption(
        Poco::Util::Option("help", "h", "show help and exit")
            .required(false)
            .repeatable(false)
            .binding("help"));
    options.addOption(
        Poco::Util::Option("version", "V", "show version and exit")
            .required(false)
            .repeatable(false)
            .binding("version"));
    BaseDaemon::defineOptions(options);
}


[[maybe_unused]] static void checkForUsersNotInMainConfig(
    const Poco::Util::AbstractConfiguration & config,
    const ServerSettings & server_settings,
    const std::string & config_path,
    const std::string & users_config_path,
    LoggerPtr log)
{
    if (server_settings[ServerSetting::skip_check_for_incorrect_settings])
        return;

    if (config.has("users") || config.has("profiles") || config.has("quotas"))
    {
        /// We cannot throw exception here, because we have support for obsolete 'conf.d' directory
        /// (that does not correspond to config.d or users.d) but substitute configuration to both of them.

        LOG_ERROR(log, "The <users>, <profiles> and <quotas> elements should be located in users config file: {} not in main config {}."
            " Also note that you should place configuration changes to the appropriate *.d directory like 'users.d'.",
            users_config_path, config_path);
    }
}

namespace
{

void loadStartupScripts(const Poco::Util::AbstractConfiguration & config, const ServerSettings & server_settings, ContextMutablePtr context, Poco::Logger * log)
{
    try
    {
        Poco::Util::AbstractConfiguration::Keys keys;
        config.keys("startup_scripts", keys);

        SetResultDetailsFunc callback;
        std::vector<String> skipped_startup_scripts;

        for (const auto & key : keys)
        {
            if (key == "throw_on_error")
                continue;
            std::string full_prefix = "startup_scripts." + key;

            auto user = config.getString(full_prefix + ".user", "");
            auto startup_context = Context::createCopy(context);

            if (!user.empty())
            {
                auto & access_control = startup_context->getAccessControl();
                startup_context->setUser(access_control.getID<User>(user));
            }

            startup_context->makeQueryContext();

            if (config.has(full_prefix + ".condition"))
            {
                auto condition = config.getString(full_prefix + ".condition");
                auto condition_read_buffer = ReadBufferFromString(condition);
                auto condition_write_buffer = WriteBufferFromOwnString();

                LOG_DEBUG(log, "Checking startup query condition `{}`", formatQueryForLogging(condition, startup_context->getSettingsRef()));
                startup_context->setQueryKind(ClientInfo::QueryKind::INITIAL_QUERY);
                startup_context->setCurrentQueryId("");

                {
                    auto query_scope = QueryScope::create(startup_context);
                    executeQuery(condition_read_buffer, condition_write_buffer, startup_context, callback, QueryFlags{ .internal = true }, std::nullopt, {});
                }

                auto result = condition_write_buffer.str();
                if (result != "1\n" && result != "true\n")
                {
                    if (result != "0\n" && result != "false\n")
                    {
                        if (result.empty())
                            LOG_DEBUG(log, "Skipping startup script as condition query returned empty value.");
                        else
                            LOG_DEBUG(
                                log,
                                "Skipping startup script as condition query returned value `{}` "
                                "which can't be interpreted as a boolean (`0`, `false`, `1`, `true`).",
                                result);
                        skipped_startup_scripts.emplace_back(full_prefix);
                    }
                    continue;
                }

                LOG_DEBUG(log, "Condition is true, will execute the query next");
            }

            auto query = config.getString(full_prefix + ".query");
            auto read_buffer = ReadBufferFromString(query);
            auto write_buffer = WriteBufferFromOwnString();

            LOG_DEBUG(log, "Executing query `{}`", formatQueryForLogging(query, startup_context->getSettingsRef()));
            startup_context->setQueryKind(ClientInfo::QueryKind::INITIAL_QUERY);
            startup_context->setCurrentQueryId("");

            auto query_scope = QueryScope::create(startup_context);

            executeQuery(read_buffer, write_buffer, startup_context, callback, QueryFlags{ .internal = true }, std::nullopt, {});
        }

        if (!skipped_startup_scripts.empty())
        {
            context->addOrUpdateWarningMessage(
                Context::WarningType::SKIPPING_CONDITION_QUERY,
                PreformattedMessage::create(
                    "Skipped the following startup script(s): {} as the condition query for those returned values, "
                    "which can't be interpreted as a boolean (`0`, `false`, `1`, `true`).",
                    fmt::join(skipped_startup_scripts, ", ")));
        }

        CurrentMetrics::set(CurrentMetrics::StartupScriptsExecutionState, StartupScriptsExecutionState::Success);
    }
    catch (...)
    {
        DimensionalMetrics::set(
            DimensionalMetrics::StartupScriptsFailureReason, {String(ErrorCodes::getName(getCurrentExceptionCode()))}, 1.0);

        CurrentMetrics::set(CurrentMetrics::StartupScriptsExecutionState, StartupScriptsExecutionState::Failure);
        tryLogCurrentException(log, "Failed to parse startup scripts file");
        if (server_settings[ServerSetting::startup_scripts_throw_on_error])
            throw Exception(
                ErrorCodes::STARTUP_SCRIPTS_ERROR,
                "Cannot finish startup_script successfully. Use startup_scripts.throw_on_error setting to change this behavior");
    }
}

void initializeAzureSDKLogger(
    [[ maybe_unused ]] const ServerSettings & server_settings,
    [[ maybe_unused ]] int server_logs_level)
{
#if USE_AZURE_BLOB_STORAGE
    if (!server_settings[ServerSetting::enable_azure_sdk_logging])
        return;

    using AzureLogsLevel = Azure::Core::Diagnostics::Logger::Level;

    static const std::unordered_map<AzureLogsLevel, std::pair<Poco::Message::Priority, DB::LogsLevel>> azure_to_server_mapping =
    {
        {AzureLogsLevel::Error, {Poco::Message::PRIO_DEBUG, LogsLevel::debug}},
        {AzureLogsLevel::Warning, {Poco::Message::PRIO_DEBUG, LogsLevel::debug}},
        {AzureLogsLevel::Informational, {Poco::Message::PRIO_TRACE, LogsLevel::trace}},
        {AzureLogsLevel::Verbose, {Poco::Message::PRIO_TEST, LogsLevel::test}},
    };

    static const std::map<Poco::Message::Priority, AzureLogsLevel> server_to_azure_mapping =
    {
        {Poco::Message::PRIO_DEBUG, AzureLogsLevel::Warning},
        {Poco::Message::PRIO_TRACE, AzureLogsLevel::Informational},
        {Poco::Message::PRIO_TEST, AzureLogsLevel::Verbose},
    };

    static const LoggerPtr azure_sdk_logger = getLogger("AzureSDK");

    auto it = server_to_azure_mapping.lower_bound(static_cast<Poco::Message::Priority>(server_logs_level));
    chassert(it != server_to_azure_mapping.end());
    Azure::Core::Diagnostics::Logger::SetLevel(it->second);

    Azure::Core::Diagnostics::Logger::SetListener([](AzureLogsLevel level, const std::string & message)
    {
        auto [poco_level, db_level] = azure_to_server_mapping.at(level);
        LOG_IMPL(azure_sdk_logger, db_level, poco_level, fmt::runtime(message));
    });
#endif
}

}

namespace
{
/// Defined next to `resolveHTTPHandlersKey` below, which they walk the `impl` chain with; declared here
/// because the configuration reload callback of `Server::main` validates a configuration with them.
Strings servedHTTPHandlersKeys(const Poco::Util::AbstractConfiguration & config);
bool hasPrometheusListener(const Poco::Util::AbstractConfiguration & config);
}

int Server::main(const std::vector<std::string> & /*args*/)
try
{
#if USE_SSH && defined(OS_LINUX)
    ::ssh::LibSSHInitializer::instance();
    ::ssh::libsshLogger::initialize();
#endif

    Stopwatch startup_watch;
    Poco::Logger * log = &logger();

    const size_t physical_server_memory = getMemoryAmount();

    LOG_INFO(
        log,
        "Available RAM: {}; logical cores: {}; used cores: {}.",
        formatReadableSizeWithBinarySuffix(physical_server_memory),
        std::thread::hardware_concurrency(),
        getNumberOfCPUCoresToUse() // on ARM processors it can show only enabled at current moment cores
    );

#if defined(__x86_64__)
    String cpu_info;
#define COLLECT_FLAG(X) \
    if (CPU::have##X()) \
    {                   \
        if (!cpu_info.empty()) \
            cpu_info += ", ";  \
        cpu_info += #X; \
    }

    CPU_ID_ENUMERATE(COLLECT_FLAG)
#undef COLLECT_FLAG

    LOG_INFO(log, "Available CPU instruction sets: {}", cpu_info);
#endif

    ServerSettings server_settings;
    server_settings.loadSettingsFromConfig(config());

    /// Fail closed if an operator is still on a legacy insert deduplication version. This build only
    /// writes the unified deduplication hash, so silently ignoring the setting would break the
    /// mixed-version deduplication contract; refuse to start and explain how to migrate. This must run
    /// after every server-settings load, not just this first one: the ZooKeeper-include reload below
    /// and the runtime config reloader re-read the settings, so a legacy value could otherwise slip in
    /// through a ZK-backed config or a `SYSTEM RELOAD CONFIG`.
    auto validate_insert_deduplication_version = [](const ServerSettings & settings)
    {
        const String dedup_version = settings[ServerSetting::insert_deduplication_version];
        if (dedup_version != "new_unified_hash")
            throw Exception(
                ErrorCodes::BAD_ARGUMENTS,
                "Server setting 'insert_deduplication_version' is set to '{}', but this version of ClickHouse "
                "supports only the unified insert deduplication hash ('new_unified_hash'). Remove the setting from "
                "the configuration (or set it to 'new_unified_hash'). To migrate from a version that used "
                "'old_separate_hashes' or 'compatible_double_hashes', first run on a release that supports "
                "'compatible_double_hashes' (writing both the legacy and unified hashes), then upgrade to this "
                "version. For replicated tables run it for at least 'replicated_deduplication_window_seconds' "
                "(one hour by default); the default windows retain the unified hashes of all inserts for that "
                "one-hour window, which is considered long enough to cover an insert retry loop. For "
                "non-replicated tables with 'non_replicated_deduplication_window' > 0 that window is count-based, "
                "not time-based, so run 'compatible_double_hashes' for at least that many inserts before "
                "upgrading; otherwise stale legacy block ids can survive the upgrade and a retried insert may be "
                "accepted as new.",
                dedup_version);
    };
    validate_insert_deduplication_version(server_settings);

#if defined(OS_LINUX)
    std::string executable_path = getExecutablePath();
    /// Remap before creating other threads to prevent crashes
    if (server_settings[ServerSetting::remap_executable])
    {
        LOG_DEBUG(log, "Will remap executable in memory.");
        /// remapExecutable rewrites the whole code segment in place; any other thread executing code during
        /// the window faults (and the signal handler's code is unmapped too, so it dies silently).
        ///
        /// Restart the async logging threads even if remapExecutable throws, so the exception unwinding
        /// through Server::main is logged by a fully running logger (a stopped channel falls back to
        /// synchronous delivery, but anything already queued waits until the threads run again).
        /// The remap exception is stashed rather than rethrown directly, so the restart happens outside the
        /// signal-blocking scope: the mask is restored first, and the logging threads get the same mask as
        /// during a normal startup.
        std::exception_ptr remap_exception;
        size_t remapped_size = 0;
#if USE_JEMALLOC
        bool jemalloc_background_threads_were_enabled = false;
#endif
        {
            /// `BaseDaemon::initializeTerminationAndSignalProcessing` has already installed the signal handlers
            /// and started the signal listener thread. Block the asynchronously delivered handled signals in this
            /// thread first, so that no new record can be queued into the signal pipe from here on. Nothing is
            /// lost: the signal stays pending for the process and is delivered once the mask is restored below.
            BlockSignalsScope block_signals(asynchronousHandledSignals());

#if USE_JEMALLOC
            /// `Jemalloc::setup` in `BaseDaemon::initialize` has already started jemalloc background threads
            /// (`jemalloc_enable_background_threads` defaults to true). They execute allocator code from the text
            /// segment, and they were created with an unblocked signal mask, so they could also take a handled
            /// signal inside the remap window. Writing `false` into the `background_thread` mallctl stops *and
            /// joins* every background thread before returning, so after this call they are fully quiesced.
            /// They are restarted below, after the remap. `max_background_threads` is a separate mallctl and is
            /// left untouched, so re-enabling restores the configured limit.
            if (Jemalloc::tryGetValue("background_thread", jemalloc_background_threads_were_enabled)
                && jemalloc_background_threads_were_enabled)
                Jemalloc::setBackgroundThreads(false);
#endif

            /// Blocking the signals is not enough by itself: a signal that arrived just before the mask was set
            /// may already have written a record into the signal pipe, and the listener thread would execute the
            /// corresponding callback at an arbitrary later moment - possibly inside the remap window. Stop and
            /// join the listener thread: it drains every record already queued (they are ordered in the pipe
            /// before the stop request) while the code is still mapped, and then exits. It is restarted below.
            stopSignalListener();

            /// The async logging threads poll rather than block, so join them for the duration and restart afterwards.
            /// This fails closed: if any logging thread cannot be joined, the exception propagates and the remap
            /// is aborted, instead of rewriting the text segment under a thread that may still execute from it.
            stopAsyncLoggingThreads();

            /// At this point the process is single-threaded again and all handled signals are blocked,
            /// so no code other than this thread can possibly run while the text segment is rewritten.
            try
            {
                remapped_size = remapExecutable();
            }
            catch (...)
            {
                remap_exception = std::current_exception();
            }
        }

#if USE_JEMALLOC
        /// Restarted outside the signal-blocking scope, so the new background threads inherit the same
        /// (unblocked) signal mask they had before the remap. Restarted even if remapExecutable threw,
        /// for the same reason the logging threads are: the allocator must be back in its configured
        /// state while the exception unwinds. `verifySetup` below cross-checks the state against the
        /// server settings, so a failure to restore would not go unnoticed.
        if (jemalloc_background_threads_were_enabled)
            Jemalloc::setBackgroundThreads(true);
#endif

        /// Fail closed if the restart itself throws: continuing with only part of the logging threads
        /// running would keep accepting messages into queues that no consumer drains. open() tears the
        /// partially opened channel back down before rethrowing, and a stopped channel delivers messages
        /// synchronously, so the exception propagating out of Server::main still reaches the log.
        startAsyncLoggingThreads();

        /// Restarted after the logging threads, so that if this throws, the exception is still logged normally.
        /// A pending signal delivered when the mask was restored above only writes a record into the signal
        /// pipe; the record waits there until the restarted listener thread picks it up, so it is not lost.
        startSignalListener();

        if (remap_exception)
            std::rethrow_exception(remap_exception);
        LOG_DEBUG(log, "The code ({}) in memory has been successfully remapped.", ReadableSize(remapped_size));
    }

    if (server_settings[ServerSetting::mlock_executable])
    {
        size_t min_physical_server_memory_to_mlock = server_settings[ServerSetting::mlock_executable_min_total_memory_amount_bytes];
        if (physical_server_memory >= min_physical_server_memory_to_mlock)
        {
            if (hasLinuxCapability(CAP_IPC_LOCK))
            {
                try
                {
                    /// Get the memory area with (current) code segment.
                    /// It's better to lock only the code segment instead of calling "mlockall",
                    /// because otherwise debug info will be also locked in memory, and it can be huge.
                    auto [addr, len] = getMappedArea(reinterpret_cast<void *>(mainEntryClickHouseServer));

                    LOG_TRACE(log, "Will do mlock to prevent executable memory from being paged out. It may take a few seconds.");
                    if (0 != mlock(addr, len))
                        LOG_WARNING(log, "Failed mlock: {}", errnoToString());
                    else
                        LOG_TRACE(log, "The memory map of clickhouse executable has been mlock'ed, total {}", ReadableSize(len));
                }
                catch (...)
                {
                    LOG_WARNING(log, "Cannot mlock: {}", getCurrentExceptionMessage(false));
                }
            }
            else
            {
                LOG_INFO(
                    log,
                    "It looks like the process has no CAP_IPC_LOCK capability, binary mlock will be disabled."
                    " It could happen due to incorrect ClickHouse package installation."
                    " You could resolve the problem manually with 'sudo setcap cap_ipc_lock=+ep {}'."
                    " Note that it will not work on 'nosuid' mounted filesystems.",
                    executable_path.empty() ? "/usr/bin/clickhouse" : executable_path);
            }
        }
        else
        {
            LOG_INFO(log, "Skip mlock for the clickhouse executable, because the total memory in the system ({}) is less than the minimum configured threshold (`mlock_executable_min_total_memory_amount_bytes` = {})",
                ReadableSize(physical_server_memory), ReadableSize(min_physical_server_memory_to_mlock));
        }
    }
#endif

    // If the startup_level is set in the config, we override the root logger level.
    // Specific loggers can still override it.
    std::string default_logger_level_config = server_settings[ServerSetting::logger_level];
    bool should_restore_default_logger_level = false;
    const auto & logger_startup_level_setting = server_settings[ServerSetting::logger_startup_level];
    if (logger_startup_level_setting.changed && !logger_startup_level_setting.value.empty())
    {
        /// Set the root logger level to the startup level.
        /// This is useful for debugging startup issues.
        /// The root logger level will be reset to the default level after the server is fully initialized.
        config().setString("logger.level", logger_startup_level_setting.value);
        Loggers::updateLevels(config(), logger());
        should_restore_default_logger_level = true;

        LOG_INFO(log, "Starting root logger in level {}", logger_startup_level_setting.value);
    }

    // If the startup_console_log_level is set in the config, we override the console logger level.
    // Specific loggers can still override it.
    bool console_log_level_was_set = config().has("logger.console_log_level");
    std::string original_console_log_level_config = config().getString("logger.console_log_level", "");
    bool should_restore_console_log_level = false;
    if (config().has("logger.startup_console_log_level") && !config().getString("logger.startup_console_log_level").empty())
    {
        /// Set the console logger level to the startup console level.
        /// This is useful for debugging startup issues.
        /// The console logger level will be reset to the default level after the server is fully initialized.
        config().setString("logger.console_log_level", config().getString("logger.startup_console_log_level"));
        Loggers::updateLevels(config(), logger());
        should_restore_console_log_level = true;

        LOG_INFO(log, "Starting console logger in level {}", config().getString("logger.startup_console_log_level"));
    }

    MainThreadStatus::getInstance();

#if USE_JEMALLOC
    Jemalloc::verifySetup(
        server_settings[ServerSetting::jemalloc_enable_global_profiler],
        server_settings[ServerSetting::jemalloc_enable_background_threads],
        server_settings[ServerSetting::jemalloc_max_background_threads_num],
        server_settings[ServerSetting::jemalloc_collect_global_profile_samples_in_trace_log],
        server_settings[ServerSetting::jemalloc_profiler_sampling_rate]);
#endif

    StackTrace::setShowAddresses(server_settings[ServerSetting::show_addresses_in_stack_traces]);

    ProfileEvents::setUserPerCPUEnabled(server_settings[ServerSetting::user_profile_events_per_cpu]);

#if USE_HDFS
    /// This will point libhdfs3 to the right location for its config.
    /// Note: this has to be done once at server initialization, because 'setenv' is not thread-safe.

    String libhdfs3_conf = server_settings[ServerSetting::hdfs_libhdfs3_conf];
    if (!libhdfs3_conf.empty())
    {
        if (std::filesystem::path{libhdfs3_conf}.is_relative() && !std::filesystem::exists(libhdfs3_conf))
        {
            const String config_path = server_settings[ServerSetting::config_file];
            const auto config_dir = std::filesystem::path{config_path}.remove_filename();
            if (std::filesystem::exists(config_dir / libhdfs3_conf))
                libhdfs3_conf = std::filesystem::absolute(config_dir / libhdfs3_conf);
        }
        setenv("LIBHDFS3_CONF", libhdfs3_conf.c_str(), true /* overwrite */); // NOLINT
    }
#endif

    if (auto total_numa_memory = getNumaNodesTotalMemory(); total_numa_memory.has_value())
    {
        LOG_INFO(
            log, "ClickHouse is bound to a subset of NUMA nodes. Total memory of all available nodes: {}", ReadableSize(*total_numa_memory));
    }

#if USE_AWS_S3
    DB::S3::setCredentialsProviderCacheMaxSize(server_settings[ServerSetting::s3_credentials_provider_max_cache_size]);
#endif

    registerInterpreters();
    registerParsers();
    registerFunctions();
    registerAggregateFunctions();
    registerTableFunctions();
    registerDatabases();
    registerStorages();
    registerDictionaries();
    setSecretArgumentsFinder(&SecretArgumentsRegistry::instance());
    registerDisks(/* global_skip_access_check= */ false);
    registerFormats();
    registerRemoteFileMetadatas();

    QueryPlanStepRegistry::registerPlanSteps();

    CurrentMetrics::set(CurrentMetrics::Revision, ClickHouseRevision::getVersionRevision());
    CurrentMetrics::set(CurrentMetrics::VersionInteger, ClickHouseRevision::getVersionInteger());

    /** Context contains all that query execution is dependent:
      *  settings, available functions, data types, aggregate functions, databases, ...
      */
    auto shared_context = Context::createShared();
    global_context = Context::createGlobal(shared_context.get());

    global_context->makeGlobalContext();
    global_context->setApplicationType(Context::ApplicationType::SERVER);

    ThreadFuzzer::instance().setup();
    addBuildWarnings(global_context);

    /// Under thread sanitizer we use frame-pointer-based unwinding (via abseil) which does not
    /// call dl_iterate_phdr in the signal handler, so the PHDR cache is not needed.
#if defined(THREAD_SANITIZER)
    bool has_trace_collector = config().has("trace_log");
    LOG_INFO(log, "Query Profiler will use frame-pointer-based stack unwinding under sanitizers.");
#else
    bool has_trace_collector = hasAsyncSignalSafeUnwind() && config().has("trace_log");
    if (!hasAsyncSignalSafeUnwind())
        LOG_INFO(log, "Query Profiler and TraceCollector are disabled because async-signal-safe stack unwinding"
            " is not available in this build (on Linux this requires the lock-free PHDR cache, otherwise"
            " 'dl_iterate_phdr' is not lock free and not async-signal safe).");
#endif

#if defined(MEMORY_SANITIZER)
    /// The TraceCollector itself stays enabled: the memory profiler, `trace_profile_events` and
    /// `SYSTEM INSTRUMENT` all feed it from ordinary code rather than from a signal handler.
    LOG_INFO(log, "The sampling Query Profiler is disabled under Memory Sanitizer, because a profiler signal"
        " delivered to a thread that is printing a sanitizer report aborts the process and truncates the"
        " report. See QUERY_PROFILER_SUPPORTED in Common/QueryProfiler.h.");
#endif

    // Settings validation for page cache. Ensure that page_cache_max_size is > page_cache_min_size.
    // Otherwise, crash might happen during cache resizing in src/Common/PageCache.cpp::autoResize
    size_t page_cache_min_size = server_settings[ServerSetting::page_cache_min_size];
    size_t page_cache_max_size = server_settings[ServerSetting::page_cache_max_size];
    if (page_cache_max_size != 0 && (page_cache_min_size > page_cache_max_size))
    {
        throw Exception(
            ErrorCodes::INVALID_CONFIG_PARAMETER,
            "Invalid page cache configuration: page_cache_min_size ({}) is greater than page_cache_max_size ({}).",
            page_cache_min_size,
            page_cache_max_size);
    }

    size_t update_period_seconds = server_settings[ServerSetting::asynchronous_metrics_update_period_s];
    size_t heavy_metrics_update_period_seconds = server_settings[ServerSetting::asynchronous_heavy_metrics_update_period_s];
    if (update_period_seconds == 0 || heavy_metrics_update_period_seconds == 0)
    {
        throw Exception(ErrorCodes::INVALID_SETTING_VALUE, "Settings asynchronous_metrics_update_period_s and asynchronous_heavy_metrics_update_period_s must not be zero");
    }

    // Initialize global thread pool. Do it before we fetch configs from zookeeper
    // nodes (`from_zk`), because ZooKeeper interface uses the pool. We will
    // ignore `max_thread_pool_size` in configs we fetch from ZK, but oh well.
    GlobalThreadPool::initialize(
        server_settings[ServerSetting::max_thread_pool_size],
        server_settings[ServerSetting::max_thread_pool_free_size],
        server_settings[ServerSetting::thread_pool_queue_size],
        has_trace_collector ? server_settings[ServerSetting::global_profiler_real_time_period_ns].value : 0,
        has_trace_collector ? server_settings[ServerSetting::global_profiler_cpu_time_period_ns].value : 0);

#if USE_SILK
    if (server_settings[ServerSetting::enable_silk_runtime])
    {
        Silk::initializeFiberScheduler(config().getUInt("silk.fiber_stack_size", Silk::DEFAULT_FIBER_STACK_SIZE));
    }
#endif

    if (has_trace_collector)
    {
        global_context->createTraceCollector();

        /// The config reloader applies this too; the seed here covers startup, which runs before its first callback.
        MemoryTracker::setMinAllocationSizeToLogStackTrace(
            server_settings[ServerSetting::min_allocation_size_to_log_stack_trace]);

        /// Set up server-wide memory profiler (for total memory tracker).
        if (server_settings[ServerSetting::total_memory_profiler_step])
            total_memory_tracker.setProfilerStep(server_settings[ServerSetting::total_memory_profiler_step]);

        if (server_settings[ServerSetting::total_memory_tracker_sample_probability] > 0.0)
            total_memory_tracker.setSampleProbability(server_settings[ServerSetting::total_memory_tracker_sample_probability]);

        if (server_settings[ServerSetting::total_memory_profiler_sample_min_allocation_size])
            total_memory_tracker.setSampleMinAllocationSize(server_settings[ServerSetting::total_memory_profiler_sample_min_allocation_size]);

        if (server_settings[ServerSetting::total_memory_profiler_sample_max_allocation_size])
            total_memory_tracker.setSampleMaxAllocationSize(server_settings[ServerSetting::total_memory_profiler_sample_max_allocation_size]);

        if (current_thread)
            current_thread->resolveMemorySampleConfig();
    }

    total_memory_tracker.setJemallocFlushProfileInterval(server_settings[ServerSetting::jemalloc_flush_profile_interval_bytes]);
    total_memory_tracker.setJemallocFlushProfileOnMemoryExceeded(server_settings[ServerSetting::jemalloc_flush_profile_on_memory_exceeded]);
    total_memory_tracker.setJemallocFlushProfileOnMemoryExceededSeconds(server_settings[ServerSetting::jemalloc_flush_profile_on_memory_exceeded_interval]);

    Poco::ThreadPool server_pool(
        /* minCapacity */3,
        /* maxCapacity */server_settings[ServerSetting::max_connections],
        /* idleTime */60,
        /* stackSize */DEFAULT_THREAD_STACK_SIZE ? static_cast<int>(DEFAULT_THREAD_STACK_SIZE) : POCO_THREAD_STACK_SIZE,
        server_settings[ServerSetting::global_profiler_real_time_period_ns],
        server_settings[ServerSetting::global_profiler_cpu_time_period_ns]);

    std::unique_ptr<Poco::ThreadPool> introspection_server_pool;

    std::mutex servers_lock;
    std::vector<ProtocolServerAdapter> servers;
    std::vector<ProtocolServerAdapter> servers_to_start_before_tables;
    std::vector<ProtocolServerAdapter> introspection_servers;

    auto stop_silk_fiber_scheduler = [&]{
#if USE_SILK
        if (server_settings[ServerSetting::enable_silk_runtime])
        {
            LOG_INFO(log, "Stopping silk fiber scheduler");
            Silk::destroyFiberScheduler();
        }
#endif
    };
    /// Wait for all threads to avoid possible use-after-free (for example logging objects can be already destroyed).
    SCOPE_EXIT_SAFE({
        Stopwatch watch;
        LOG_INFO(log, "Waiting for background threads");
        DB::StaticThreadPool::shutdownAll();
        GlobalThreadPool::instance().shutdown();
        LOG_INFO(log, "Background threads finished in {} ms", watch.elapsedMilliseconds());

        stop_silk_fiber_scheduler();
    });

    if (page_cache_max_size != 0)
    {
        global_context->setPageCache(
            std::chrono::milliseconds(Int64(server_settings[ServerSetting::page_cache_history_window_ms])),
            server_settings[ServerSetting::page_cache_policy],
            server_settings[ServerSetting::page_cache_size_ratio],
            server_settings[ServerSetting::page_cache_min_size],
            server_settings[ServerSetting::page_cache_max_size],
            server_settings[ServerSetting::page_cache_free_memory_ratio],
            server_settings[ServerSetting::page_cache_shards]);
        total_memory_tracker.setPageCache(global_context->getPageCache().get());
    }

    MemoryWorkerConfig memory_worker_config{
        .rss_update_period_ms = server_settings[ServerSetting::memory_worker_period_ms],
        .purge_dirty_pages_threshold_ratio = server_settings[ServerSetting::memory_worker_purge_dirty_pages_threshold_ratio],
        .purge_total_memory_threshold_ratio = server_settings[ServerSetting::memory_worker_purge_total_memory_threshold_ratio],
        .correct_tracker = server_settings[ServerSetting::memory_worker_correct_memory_tracker],
        .decay_adjustment_period_ms = server_settings[ServerSetting::memory_worker_decay_adjustment_period_ms],
        .use_cgroup = server_settings[ServerSetting::memory_worker_use_cgroup],
        .rss_speculative_reserve_ratio = server_settings[ServerSetting::memory_worker_rss_speculative_reserve_ratio],
        .dynamic_hard_limit_ratio = server_settings[ServerSetting::memory_worker_dynamic_hard_limit]
            ? static_cast<double>(server_settings[ServerSetting::max_server_memory_usage_to_ram_ratio])
            : 0.0,
    };

    MemoryWorker memory_worker(memory_worker_config, global_context->getPageCache());

    /// This object will periodically calculate some metrics.
    std::unique_ptr<AsynchronousMetrics> async_metrics;

    DB::AsynchronousMetrics::ProtocolServerMetricsFunc asynchronous_metrics_protocol_server_metrics_func = [&]() -> std::vector<ProtocolServerMetrics>
    {
        std::vector<ProtocolServerMetrics> metrics;

        std::lock_guard lock(servers_lock);
        metrics.reserve(servers_to_start_before_tables.size() + servers.size() + introspection_servers.size());

        for (const auto & server : servers_to_start_before_tables)
            metrics.emplace_back(ProtocolServerMetrics{
                server.getPortName(), server.getProtocolType(), server.currentThreads(), server.refusedConnections()});

        for (const auto & server : servers)
            metrics.emplace_back(ProtocolServerMetrics{
                server.getPortName(), server.getProtocolType(), server.currentThreads(), server.refusedConnections()});

        for (const auto & server : introspection_servers)
            metrics.emplace_back(ProtocolServerMetrics{
                server.getPortName(), server.getProtocolType(), server.currentThreads(), server.refusedConnections()});
        return metrics;
    };
    const unsigned async_metrics_update_period_s = server_settings[ServerSetting::asynchronous_metrics_update_period_s];
    const bool async_metrics_update_jemalloc_epoch = memory_worker.getSource() != MemoryWorker::MemoryUsageSource::Jemalloc;
    const bool async_metrics_update_rss = memory_worker.getSource() == MemoryWorker::MemoryUsageSource::None;

#if USE_NURAFT
    const bool async_metrics_keeper_metrics_only = server_settings[ServerSetting::asynchronous_metrics_keeper_metrics_only];
    if (async_metrics_keeper_metrics_only)
    {
        async_metrics = std::make_unique<KeeperAsynchronousMetrics>(
            global_context,
            async_metrics_update_period_s,
            asynchronous_metrics_protocol_server_metrics_func,
            async_metrics_update_jemalloc_epoch,
            async_metrics_update_rss
        );
    }
    else
#endif
    {
        async_metrics = std::make_unique<ServerAsynchronousMetrics>(
            global_context,
            async_metrics_update_period_s,
            server_settings[ServerSetting::asynchronous_metrics_enable_heavy_metrics],
            server_settings[ServerSetting::asynchronous_heavy_metrics_update_period_s],
            asynchronous_metrics_protocol_server_metrics_func,
            async_metrics_update_jemalloc_epoch,
            async_metrics_update_rss
        );
    }

    auto begin_keeper_tcp_connection_drain = [&]
    {
#if USE_NURAFT
        if (auto dispatcher = global_context->tryGetKeeperDispatcher())
            dispatcher->beginTCPConnectionDrain();

        KeeperTCPHandler::closeAllConnections();
#endif
    };

    auto is_keeper_tcp_server = [](const ProtocolServerAdapter & server)
    {
        const auto & port_name = server.getPortName();
        return port_name == "keeper_server.tcp_port" || port_name == "keeper_server.tcp_port_secure";
    };

    /// NOTE: global context should be destroyed *before* GlobalThreadPool::shutdown()
    /// Otherwise GlobalThreadPool::shutdown() will hang, since Context holds some threads.
    SCOPE_EXIT_SAFE({
        /// Required for the startup exception case where the termination block that should set is_cancelled=true never runs.
        is_cancelled = true;

        /// Stop accepting connections on the regular servers. In the normal shutdown path they are
        /// already stopped, but on startup failure some of them can still be running: the Prometheus
        /// endpoint is started before tables are loaded. Otherwise `server_pool.joinAll()` below
        /// would wait forever for their listener threads.
        {
            std::lock_guard lock(servers_lock);
            for (auto & server : servers)
                if (!server.isStopping())
                    server.stop();
        }

        async_metrics->stop();

        /** Ask to cancel background jobs all table engines,
          *  and also query_log.
          * It is important to do early, not in destructor of Context, because
          *  table engines could use Context on destroy.
          */
        LOG_INFO(log, "Shutting down storages.");

        /// The columns cache goes away with the context; stop resizing it.
        setMemoryReleasableCache(nullptr);
        memory_worker.setReleasableCache(nullptr);

        global_context->shutdown();

        LOG_DEBUG(log, "Shut down storages.");
        size_t keeper_tcp_connections = 0;
        size_t non_keeper_tcp_connections = 0;
        if (!servers_to_start_before_tables.empty())
        {
            LOG_DEBUG(log, "Waiting for current connections to servers for tables to finish.");
            {
                std::lock_guard lock(servers_lock);
                for (auto & server : servers_to_start_before_tables)
                {
                    server.stop();
                    if (is_keeper_tcp_server(server))
                        keeper_tcp_connections += server.currentConnections();
                    else
                        non_keeper_tcp_connections += server.currentConnections();
                }
            }
        }

        /// Stop Keeper TCP handlers before draining the remaining pre-table protocol handlers.
        /// Those handlers include HTTP control, interserver HTTP/HTTPS, and Prometheus; they need
        /// the embedded Keeper and RAFT to remain live until they finish.
        begin_keeper_tcp_connection_drain();

        if (non_keeper_tcp_connections)
        {
            LOG_INFO(log, "Closed all non-Keeper-TCP listening sockets. Waiting for {} outstanding connections.", non_keeper_tcp_connections);
            non_keeper_tcp_connections = waitServersToFinish(
                servers_to_start_before_tables,
                servers_lock,
                server_settings[ServerSetting::shutdown_wait_unfinished],
                [&](const auto & server) { return !is_keeper_tcp_server(server); });
        }

        global_context->signalKeeperDispatcherShutdown();
        global_context->shutdownKeeperDispatcherBeforeConnectionsFinish();

        if (non_keeper_tcp_connections)
        {
            global_context->shutdownKeeperDispatcherAfterConnectionsFinish(false);
            dumpCoverageReportIfPossible();
            LOG_WARNING(
                log,
                "Closed connections to non-Keeper-TCP servers. But {} remain. Will shutdown forcefully.",
                non_keeper_tcp_connections);
            safeExit(0, LeakCheck::SkipAndReport);
        }

        if (!servers_to_start_before_tables.empty())
        {
            if (keeper_tcp_connections)
                LOG_INFO(log, "Closed all Keeper TCP listening sockets. Waiting for {} outstanding connections.", keeper_tcp_connections);
            else
                LOG_INFO(log, "Closed all Keeper listening sockets.");

            if (keeper_tcp_connections > 0)
                keeper_tcp_connections = waitServersToFinish(
                    servers_to_start_before_tables,
                    servers_lock,
                    server_settings[ServerSetting::shutdown_wait_unfinished],
                    is_keeper_tcp_server);

            if (keeper_tcp_connections)
                LOG_INFO(log, "Closed Keeper TCP connections. But {} remain.", keeper_tcp_connections);
            else
                LOG_INFO(log, "Closed Keeper TCP connections.");
        }

        global_context->shutdownKeeperDispatcherAfterConnectionsFinish(keeper_tcp_connections == 0);

        /// Wait server pool to avoid use-after-free of destroyed context in the handlers
        server_pool.joinAll();

        /** Explicitly destroy Context. It is more convenient than in destructor of Server, because logger is still available.
          * At this moment, no one could own shared part of Context.
          */
        global_context->resetSharedContext();
        global_context.reset();
        shared_context.reset();
        LOG_DEBUG(log, "Destroyed global context.");
    });


#if USE_AZURE_BLOB_STORAGE
    /// It makes sense to deinitialize libxml after joining of all threads
    /// in global pool because libxml uses thread-local memory allocations via
    /// 'pthread_key_create' and 'pthread_setspecific' which should be deallocated
    /// at 'pthread_exit'. Deinitialization of libxml leads to call of 'pthread_key_delete'
    /// and if it is done before joining of threads, allocated memory will not be freed
    /// and there may be memory leaks in threads that used libxml.
    GlobalThreadPool::instance().addOnDestroyCallback([]
    {
        Azure::Storage::_internal::XmlGlobalDeinitialize();
    });
#endif

    getIOThreadPool().initialize(
        server_settings[ServerSetting::max_io_thread_pool_size],
        server_settings[ServerSetting::max_io_thread_pool_free_size],
        server_settings[ServerSetting::io_thread_pool_queue_size]);

    getBackupsIOThreadPool().initialize(
        server_settings[ServerSetting::max_backups_io_thread_pool_size],
        server_settings[ServerSetting::max_backups_io_thread_pool_free_size],
        server_settings[ServerSetting::backups_io_thread_pool_queue_size]);

    getFetchPartitionThreadPool().initialize(
        server_settings[ServerSetting::max_fetch_partition_thread_pool_size],
        0, // FETCH PARTITION is relatively rare, no need to keep threads
        server_settings[ServerSetting::max_fetch_partition_thread_pool_size]);

    getActivePartsLoadingThreadPool().initialize(
        server_settings[ServerSetting::max_active_parts_loading_thread_pool_size],
        0, // We don't need any threads once all the parts will be loaded
        server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]);

    getOutdatedPartsLoadingThreadPool().initialize(
        server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size],
        0, // We don't need any threads once all the parts will be loaded
        server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size]);

    /// It could grow if we need to synchronously wait until all the data parts will be loaded.
    getOutdatedPartsLoadingThreadPool().setMaxTurboThreads(
        server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]
    );

    getUnexpectedPartsLoadingThreadPool().initialize(
        server_settings[ServerSetting::max_unexpected_parts_loading_thread_pool_size],
        0, // We don't need any threads once all the parts will be loaded
        server_settings[ServerSetting::max_unexpected_parts_loading_thread_pool_size]);

    /// It could grow if we need to synchronously wait until all the data parts will be loaded.
    getUnexpectedPartsLoadingThreadPool().setMaxTurboThreads(
        server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]
    );

    getPartsCleaningThreadPool().initialize(
        server_settings[ServerSetting::max_parts_cleaning_thread_pool_size],
        0, // We don't need any threads one all the parts will be deleted
        server_settings[ServerSetting::max_parts_cleaning_thread_pool_size]);

    auto max_database_replicated_create_table_thread_pool_size = server_settings[ServerSetting::max_database_replicated_create_table_thread_pool_size]
        ? server_settings[ServerSetting::max_database_replicated_create_table_thread_pool_size]
        : getNumberOfCPUCoresToUse();
    getDatabaseReplicatedCreateTablesThreadPool().initialize(
        max_database_replicated_create_table_thread_pool_size,
        0, // We don't need any threads once all the tables will be created
        max_database_replicated_create_table_thread_pool_size);

    getDatabaseCatalogDropTablesThreadPool().initialize(
        server_settings[ServerSetting::database_catalog_drop_table_concurrency],
        0, // We don't need any threads if there are no DROP queries.
        server_settings[ServerSetting::database_catalog_drop_table_concurrency]);

    /// Zero means the number of CPU cores.
    const size_t shutdown_concurrency = server_settings[ServerSetting::database_catalog_shutdown_table_concurrency]
        ? server_settings[ServerSetting::database_catalog_shutdown_table_concurrency]
        : getNumberOfCPUCoresToUse();
    getDatabaseCatalogShutdownTablesThreadPool().initialize(
        shutdown_concurrency,
        0, // Threads are only needed during server shutdown.
        shutdown_concurrency);

    getMergeTreePrefixesDeserializationThreadPool().initialize(
        server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_size],
        server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_free_size],
        server_settings[ServerSetting::prefixes_deserialization_thread_pool_thread_pool_queue_size]);

    getFormatParsingThreadPool().initialize(
        server_settings[ServerSetting::max_format_parsing_thread_pool_size],
        server_settings[ServerSetting::max_format_parsing_thread_pool_free_size],
        server_settings[ServerSetting::format_parsing_thread_pool_queue_size]);

    getAIRequestThreadPool().initialize(
        server_settings[ServerSetting::ai_function_max_thread_pool_size],
        server_settings[ServerSetting::ai_function_max_thread_pool_free_size],
        server_settings[ServerSetting::ai_function_thread_pool_queue_size]);

    std::string path_str = getCanonicalPath(String(server_settings[ServerSetting::path]), original_working_directory);
    fs::path path = path_str;

    /// Check that the process user id matches the owner of the data.
    assertProcessUserMatchesDataOwner(
        path_str,
        [&](const PreformattedMessage & message)
        {
            global_context->addOrUpdateWarningMessage(
                Context::WarningType::PROCESS_USER_MATCHES_DATA_OWNER, message);
        });

    global_context->setPath(path_str);

    StatusFile status{path / "status", StatusFile::write_full_info};

    ServerUUID::load(path / "uuid", log);

    PlacementInfo::PlacementInfo::instance().initialize(config());

    zkutil::validateZooKeeperConfig(config());
    bool has_zookeeper = zkutil::hasZooKeeperConfig(config());

    if (has_zookeeper && config().getBool("keeper_server.standalone_keeper", false))
        throw Exception(ErrorCodes::BAD_ARGUMENTS,
            "Cannot use standalone_keeper=true with a configured zookeeper client connection. "
            "A standalone keeper should not have a <zookeeper> section in its config.");

    auto main_config_zk_node_cache = std::make_unique<zkutil::ZooKeeperNodeCache>([&] { return global_context->getZooKeeper(); });
    Coordination::EventPtr main_config_zk_changed_event = std::make_shared<Poco::Event>();
    if (loaded_config.has_zk_includes)
    {
        ConfigProcessor config_processor(config_path);
        loaded_config = config_processor.loadConfigWithZooKeeperIncludes(
            main_config_zk_node_cache.get(), main_config_zk_changed_event, /* fallback_to_preprocessed = */ true);
        config_processor.savePreprocessedConfig(loaded_config, path_str);
        config().replace("default", loaded_config.configuration.duplicate(), PRIO_DEFAULT, false);
    }

    Settings::checkNoSettingNamesAtTopLevel(config(), config_path);
    /// Validate the loaded XML config directly (not the layered config) so that command-line
    /// options injected by `argsToConfig` (`--config-file`, `--daemon`, `-C`, ...) and Poco-internal
    /// layers (`system`, `application`) are not mistaken for unknown top-level config keys.
    /// The `skip_check_for_incorrect_settings` escape hatch is resolved from the layered `config()`,
    /// so it works when supplied from the command line as well as from the config file.
    ServerSettings::checkUnknownSettings(
        *loaded_config.configuration, config_path, config().getBool("skip_check_for_incorrect_settings", false));

    /// We need to reload server settings because config could be updated via zookeeper.
    server_settings.loadSettingsFromConfig(config());
    validate_insert_deduplication_version(server_settings);
    global_context->configureServerWideThrottling();

    /// Create the dedicated MergeTree metadata arena pool. Placed after the ZooKeeper-include reload
    /// above so a `from_zk` value of the setting is honored, and still well before any parts are loaded.
    JemallocMergeTreeArena::initialize(server_settings[ServerSetting::jemalloc_merge_tree_arenas]);
    addMergeTreeArenaPoolWarnings(global_context);

    /// Restrict the server to the system calls it is known to use, as early in the startup as the
    /// configuration allows. That is after the ZooKeeper-include reload above, not before it: a
    /// filter cannot be removed or relaxed afterwards, so one installed from the configuration as it
    /// was before a `from_zk` value arrived would stay in force while `system.server_settings`
    /// reported the value from ZooKeeper. `TSYNC` extends the filter to the threads that already
    /// exist, so everything that runs from here on - including the processes the server forks
    /// later, which inherit it - is covered.
    const SeccompMode seccomp_mode = server_settings[ServerSetting::seccomp];
    if (const SeccompFilterStatus seccomp_status = installSeccompFilter(seccomp_mode); seccomp_status.allowed_syscalls != 0)
        LOG_INFO(
            log,
            "Applied a seccomp policy to this process, allowing {} system calls. A system call outside the policy will "
            "be handled according to the `seccomp` server setting, which is set to `{}`",
            seccomp_status.allowed_syscalls,
            SettingFieldSeccompMode(seccomp_mode).toString());
    else if (seccomp_mode != SeccompMode::Disabled)
#if defined(OS_LINUX)
        LOG_WARNING(
            log,
            "The `seccomp` server setting is set to `{}`, but {}, so the server is running without a seccomp policy. "
            "`PR_SET_NO_NEW_PRIVS` has been set anyway, so nothing this process runs can gain privileges through a setuid program",
            SettingFieldSeccompMode(seccomp_mode).toString(),
            seccomp_status.not_installed_reason);
#else
        LOG_WARNING(
            log,
            "The `seccomp` server setting is set to `{}`, but {}, so the server is running without a seccomp policy",
            SettingFieldSeccompMode(seccomp_mode).toString(),
            seccomp_status.not_installed_reason);
#endif

#if defined(OS_LINUX)
    if (server_settings[ServerSetting::skip_binary_checksum_checks])
    {
        LOG_WARNING(log, "Binary checksum checks disabled due to skip_binary_checksum_checks - not recommended for production deployments");
    }
    else if (!executable_path.empty())
    {
        /// Integrity check based on checksum of the executable code.
        /// Note: it is not intended to protect from malicious party,
        /// because the reference checksum can be easily modified as well.
        /// And we don't involve asymmetric encryption with PKI yet.
        /// It's only intended to protect from faulty hardware.
        /// Note: it is only based on machine code.
        /// But there are other sections of the binary (e.g. exception handling tables)
        /// that are interpreted (not executed) but can alter the behaviour of the program as well.

        /// Please keep the below log messages in-sync with the ones in daemon/BaseDaemon.cpp
        if (stored_binary_hash.empty())
        {
            LOG_WARNING(log, "Integrity check of the executable skipped because the reference checksum could not be read.");
        }
        else
        {
            String calculated_binary_hash = getHashOfLoadedBinaryHex();
            if (calculated_binary_hash == stored_binary_hash)
            {
                LOG_INFO(log, "Integrity check of the executable successfully passed (checksum: {})", calculated_binary_hash);
            }
            else
            {
                /// Note: this must not rely on a system call the `seccomp` policy denies, such as
                /// `ptrace`, because the filter is already installed by this point.
                if (isRunUnderDebugger())
                {
                    /// Program is run under debugger. Modification of it's binary image is ok for breakpoints.
                    global_context->addOrUpdateWarningMessage(
                        Context::WarningType::SERVER_RUN_UNDER_DEBUGGER,
                        PreformattedMessage::create(
                            "Server is run under debugger and its binary image {} is modified (most likely with breakpoints).",
                            calculated_binary_hash));
                }
                else
                {
                    throw Exception(
                        ErrorCodes::CORRUPTED_DATA,
                        "Calculated checksum of the executable ({0}) does not correspond"
                        " to the reference checksum stored in the executable ({1})."
                        " This may indicate one of the following:"
                        " - the executable {2} was changed just after startup;"
                        " - the executable {2} was corrupted on disk due to faulty hardware;"
                        " - the loaded executable was corrupted in memory due to faulty hardware;"
                        " - the file {2} was intentionally modified;"
                        " - a logical error in the code.",
                        calculated_binary_hash,
                        stored_binary_hash,
                        executable_path);
                }
            }
        }
    }

    /// Inject and enable failpoints declared in the configuration
    {
        auto & conf = config();
        String root_key = "fail_points_active";
        Poco::Util::AbstractConfiguration::Keys fail_point_names;
        conf.keys(root_key, fail_point_names);

        for (const auto & fail_point_name : fail_point_names)
        {
            if (ConfigHelper::getBool(conf, root_key + "." + fail_point_name))
                FailPointInjection::enableFailPoint(fail_point_name);
        }
    }

#endif

    memory_worker.start();

#if defined(OS_LINUX)
    int oom_score = server_settings[ServerSetting::oom_score];
    if (oom_score)
        setOOMScore(oom_score, log);
#endif

#if defined(OS_LINUX)
    std::optional<OOMCanary> oom_canary;
    if (server_settings[ServerSetting::oom_canary_enable])
    {
        OOMCanary::Config canary_config;
        canary_config.size_bytes = server_settings[ServerSetting::oom_canary_size];
        canary_config.relaunch = server_settings[ServerSetting::oom_canary_relaunch];
        canary_config.max_rapid_relaunches = server_settings[ServerSetting::oom_canary_max_rapid_relaunches];
        canary_config.initial_backoff_seconds = server_settings[ServerSetting::oom_canary_initial_backoff_seconds];
        canary_config.max_backoff_seconds = server_settings[ServerSetting::oom_canary_max_backoff_seconds];
        oom_canary.emplace(global_context, std::move(canary_config));
    }
    else
    {
        LOG_INFO(log, "OOM canary is disabled");
    }
#else
    if (server_settings[ServerSetting::oom_canary_enable])
        LOG_WARNING(log, "OOM canary is only supported on Linux, ignoring");
#endif

    std::unique_ptr<DB::BackgroundSchedulePoolTaskHolder> cancellation_task;

    SCOPE_EXIT({
        if (cancellation_task)
            CancellationChecker::getInstance().terminateThread();
    });

    if (server_settings[ServerSetting::background_schedule_pool_size] > 1)
    {
        auto cancellation_task_holder = global_context->getSchedulePool()->createTask(
            StorageID::createEmpty(), "CancellationChecker",
            [] { CancellationChecker::getInstance().workerFunction(); }
        );
        cancellation_task = std::make_unique<DB::BackgroundSchedulePoolTaskHolder>(std::move(cancellation_task_holder));
        (*cancellation_task)->activateAndSchedule();
    }

    global_context->setRemoteHostFilter(config());
    global_context->setHTTPHeaderFilter(config());

    /// Try to increase limit on number of open files.
    {
        rlimit rlim{};
        if (getrlimit(RLIMIT_NOFILE, &rlim))
            throw Poco::Exception("Cannot getrlimit");

        if (rlim.rlim_cur == rlim.rlim_max)
        {
            LOG_DEBUG(log, "rlimit on number of file descriptors is {}", rlim.rlim_cur);
        }
        else
        {
            rlim_t old = rlim.rlim_cur;
            rlim.rlim_cur = server_settings[ServerSetting::max_open_files].changed ? static_cast<rlim_t>(server_settings[ServerSetting::max_open_files]) : rlim.rlim_max;
            int rc = setrlimit(RLIMIT_NOFILE, &rlim);
            if (rc != 0)
                LOG_WARNING(log, "Cannot set max number of file descriptors to {}. Try to specify max_open_files according to your system limits. error: {}", rlim.rlim_cur, errnoToString());
            else
                LOG_DEBUG(log, "Set max number of file descriptors to {} (was {}).", rlim.rlim_cur, old);
        }
    }

#if defined(RLIMIT_NPROC)
    /// Try to increase limit on number of threads.
    {
        rlimit rlim{};
        if (getrlimit(RLIMIT_NPROC, &rlim))
            throw Poco::Exception("Cannot getrlimit");

        if (rlim.rlim_cur == rlim.rlim_max)
        {
            LOG_DEBUG(log, "rlimit on number of threads is {}", rlim.rlim_cur);
        }
        else
        {
            rlim_t old = rlim.rlim_cur;
            rlim.rlim_cur = rlim.rlim_max;
            int rc = setrlimit(RLIMIT_NPROC, &rlim);
            if (rc != 0)
            {
                LOG_WARNING(log, "Cannot set max number of threads to {}. error: {}", rlim.rlim_cur, errnoToString());
                rlim.rlim_cur = old;
            }
            else
            {
                LOG_DEBUG(log, "Set max number of threads to {} (was {}).", rlim.rlim_cur, old);
            }
        }

        if (rlim.rlim_cur < 30000)
        {
            global_context->addOrUpdateWarningMessage(
                Context::WarningType::MAX_NUM_THREADS_LOWER_THAN_LIMIT,
                PreformattedMessage::create(
                    "Maximum number of threads is lower than 30000. There could be problems with handling a lot of simultaneous queries."));
        }
    }
#endif

    static ServerErrorHandler error_handler;
    Poco::ErrorHandler::set(&error_handler);

    /// Initialize DateLUT early, to not interfere with running time of first query.
    LOG_DEBUG(log, "Initializing DateLUT.");
    DateLUT::serverTimezoneInstance();
    LOG_TRACE(log, "Initialized DateLUT with time zone '{}'.", getDateLUTTimeZone(DateLUT::serverTimezoneInstance()));

    if (config().has("macros"))
        global_context->setMacros(std::make_unique<Macros>(config(), "macros", log));

    /** Directory with 'flags': files indicating temporary settings for the server set by system administrator.
      * Flags may be cleared automatically after being applied by the server.
      * Examples: do repair of local data; clone all replicated tables from replica.
      */
    {
        auto flags_path = path / "flags/";
        fs::create_directories(flags_path);
        global_context->setFlagsPath(flags_path);
    }

    /** Directory with user provided files that are usable by 'file' table function.
      */
    {
        const auto & user_files_path_setting = server_settings[ServerSetting::user_files_path];
        std::string user_files_path = user_files_path_setting.changed
            ? getCanonicalPath(String(user_files_path_setting.value), path_str) : String(path / "user_files/");
        global_context->setUserFilesPath(user_files_path);
        fs::create_directories(user_files_path);
    }

    {
        const auto & user_scripts_path_setting = server_settings[ServerSetting::user_scripts_path];
        std::string user_scripts_path = user_scripts_path_setting.changed
            ? getCanonicalPath(String(user_scripts_path_setting.value), path_str) : String(path / "user_scripts/");
        global_context->setUserScriptsPath(user_scripts_path);
    }

    {
        const auto & dynamic_udf_path_setting = server_settings[ServerSetting::dynamic_user_defined_executable_functions_path];
        std::string dynamic_udf_path = dynamic_udf_path_setting.changed
            ? getCanonicalPath(String(dynamic_udf_path_setting.value), path_str) : String(path / "dynamic_user_defined_executable_functions/");
        global_context->setDynamicUserDefinedExecutableFunctionsPath(dynamic_udf_path);
        fs::create_directories(dynamic_udf_path);
    }

    /// top_level_domains_lists
    {
        const auto & top_level_domains_path_setting = server_settings[ServerSetting::top_level_domains_path];
        std::string top_level_domains_path = top_level_domains_path_setting.changed
            ? getCanonicalPath(String(top_level_domains_path_setting.value), path_str) : String(path / "top_level_domains/");
        TLDListsHolder::getInstance().parseConfig(fs::path(top_level_domains_path) / "", config());
    }

    {
        fs::create_directories(path / "data");
        fs::create_directories(path / "metadata");

        /// Directory with metadata of tables, which was marked as dropped by Atomic database
        fs::create_directories(path / "metadata_dropped");
    }

    if (server_settings[ServerSetting::interserver_http_port].changed && server_settings[ServerSetting::interserver_https_port].changed)
        throw Exception(ErrorCodes::EXCESSIVE_ELEMENT_IN_CONFIG, "Both http and https interserver ports are specified");

    static const auto interserver_tags =
    {
        std::make_tuple("interserver_http_host", server_settings[ServerSetting::interserver_http_host], "interserver_http_port", server_settings[ServerSetting::interserver_http_port], "http"),
        std::make_tuple("interserver_https_host", server_settings[ServerSetting::interserver_https_host], "interserver_https_port", server_settings[ServerSetting::interserver_https_port], "https")
    };

    for (auto [host_tag, host_setting, port_tag, port_setting, scheme] : interserver_tags)
    {
        if (port_setting.changed)
        {
            String this_host = host_setting.value;

            if (this_host.empty())
            {
                this_host = getFQDNOrHostName();
                LOG_DEBUG(log, "Configuration parameter '{}' doesn't exist or exists and empty. Will use '{}' as replica host.",
                    host_tag, this_host);
            }

            UInt64 port = port_setting.value;

            if (port > 0xFFFF)
                throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "Out of range '{}': {}", String(port_tag), port);

            global_context->setInterserverIOAddress(this_host, static_cast<UInt16>(port));
            global_context->setInterserverScheme(scheme);
        }
    }

    LOG_DEBUG(log, "Initializing interserver credentials.");
    global_context->updateInterserverCredentials(config());

    std::shared_ptr<StatelessWorkerEndpoint> stateless_worker_endpoint_ptr{nullptr};
    String stateless_worker_endpoint_name;
    if (config().getBool("stateless_worker_server.enabled", false))
    {
        String stateless_worker_endpoint = config().getString("stateless_worker_server.endpoint", "localhost");
        size_t stateless_worker_max_threads = config().getUInt64("stateless_worker_server.max_threads", 1000);
        if (stateless_worker_max_threads == 0)
            throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER,
                "`stateless_worker_server.max_threads` must be at least 1");
        stateless_worker_endpoint_ptr = std::make_shared<StatelessWorkerEndpoint>(
            stateless_worker_max_threads,
            config().getUInt64("stateless_worker_server.max_free_threads", 0),
            config().getUInt64("stateless_worker_server.queue_size", 3000));
        stateless_worker_endpoint_name = stateless_worker_endpoint_ptr->getId(stateless_worker_endpoint);
        global_context->getInterserverIOHandler().addEndpoint(stateless_worker_endpoint_name, stateless_worker_endpoint_ptr);
        LOG_DEBUG(log, "Added stateless worker endpoint '{}'.", stateless_worker_endpoint_name);
    }

    SCOPE_EXIT({
        if (stateless_worker_endpoint_ptr)
        {
            /// Remove the same endpoint that was registered (the configured name may differ from "localhost").
            LOG_DEBUG(log, "Shutting down stateless worker endpoint '{}'.", stateless_worker_endpoint_name);
            global_context->getInterserverIOHandler().removeEndpointIfExists(stateless_worker_endpoint_name);

            stateless_worker_endpoint_ptr->blocker.cancelForever();
            stateless_worker_endpoint_ptr->shutdown();
            /// Acquire the lock to wait for all in-flight requests to finish.
            std::lock_guard lock(stateless_worker_endpoint_ptr->rwlock);
        }
        stateless_worker_endpoint_ptr.reset();
    });

    #if defined(OS_LINUX) || defined(OS_DARWIN)
    ExchangeConnectionsPtr exchange_connections_ptr = ExchangeConnections::instance();
    std::vector<std::shared_ptr<ExchangeServer>> exchange_servers;
    if (auto streaming_exchange_port = config().getUInt("distributed_query.streaming_exchange_port", 0))
    {
        if (streaming_exchange_port > 65535)
            throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER,
                "`distributed_query.streaming_exchange_port` must be in range 1..65535, got {}", streaming_exchange_port);

        /// The exchange handshake is unauthenticated, so the listener is never bound to all interfaces
        /// implicitly: the streaming exchange is enabled only when explicit listen host(s) are given.
        Strings exchange_listen_hosts = DB::getMultipleValuesFromConfig(config(), "distributed_query", "streaming_exchange_listen_host");
        if (exchange_listen_hosts.empty())
        {
            LOG_ERROR(log, "`distributed_query.streaming_exchange_port` is set but no "
                "`distributed_query.streaming_exchange_listen_host` is configured; the streaming exchange "
                "server is not started. Specify a listen host to enable it.");
        }
        else
        {
            for (const auto & listen_host : exchange_listen_hosts)
            {
                try
                {
                    exchange_servers.emplace_back(std::make_shared<ExchangeServer>(listen_host, streaming_exchange_port, exchange_connections_ptr));
                    exchange_servers.back()->start();
                }
                catch (Poco::Exception & e)
                {
                    LOG_INFO(log, "Failed to start exchange server on {}:{}: {}",
                        listen_host, streaming_exchange_port, e.displayText());
                }
            }
            if (exchange_servers.empty())
                throw Exception(ErrorCodes::NETWORK_ERROR, "Failed to start ExchangeServer on port {}", streaming_exchange_port);
        }
    }

    SCOPE_EXIT({
        for (auto & exchange_server_ptr : exchange_servers)
        {
            exchange_server_ptr->stop();
            exchange_server_ptr.reset();
        }
    });
    #else
    if (config().getUInt("distributed_query.streaming_exchange_port", 0))
        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "ExchangeServer is not supported on this platform");
    #endif

    /// Set up caches.

    JemallocCacheArena::setEnabled(server_settings[ServerSetting::use_separate_cache_arena]);

    const size_t max_cache_size = static_cast<size_t>(static_cast<double>(physical_server_memory) * server_settings[ServerSetting::cache_size_to_ram_max_ratio]);

    String uncompressed_cache_policy = server_settings[ServerSetting::uncompressed_cache_policy];
    size_t uncompressed_cache_size = server_settings[ServerSetting::uncompressed_cache_size];
    double uncompressed_cache_size_ratio = server_settings[ServerSetting::uncompressed_cache_size_ratio];
    if (uncompressed_cache_size > max_cache_size)
    {
        uncompressed_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered uncompressed cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(uncompressed_cache_size));
    }
    global_context->setUncompressedCache(uncompressed_cache_policy, uncompressed_cache_size, uncompressed_cache_size_ratio);

    if (server_settings[ServerSetting::enable_gpu])
    {
#if USE_GPU
        global_context->setGPU(server_settings);
#else
        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "The server setting `enable_gpu` is set, but the server is built without GPU support");
#endif
    }

    String mark_cache_policy = server_settings[ServerSetting::mark_cache_policy];
    size_t mark_cache_size = server_settings[ServerSetting::mark_cache_size];
    double mark_cache_size_ratio = server_settings[ServerSetting::mark_cache_size_ratio];
    if (mark_cache_size > max_cache_size)
    {
        mark_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered mark cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(mark_cache_size));
    }
    global_context->setMarkCache(mark_cache_policy, mark_cache_size, mark_cache_size_ratio);

    String unique_key_index_cache_policy_name = server_settings[ServerSetting::unique_key_index_cache_policy];
    size_t unique_key_index_cache_size = server_settings[ServerSetting::unique_key_index_cache_size_bytes];
    double unique_key_index_cache_size_ratio = server_settings[ServerSetting::unique_key_index_cache_size_ratio];
    if (unique_key_index_cache_size > max_cache_size)
    {
        unique_key_index_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered UNIQUE KEY index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(unique_key_index_cache_size));
    }
    global_context->setUniqueKeyIndexCache(unique_key_index_cache_policy_name, unique_key_index_cache_size, unique_key_index_cache_size_ratio);

    /// UNIQUE KEY delete-bitmap cache. Zero size disables.
    String unique_key_bitmap_cache_policy_name = server_settings[ServerSetting::unique_key_bitmap_cache_policy];
    size_t unique_key_bitmap_cache_size = server_settings[ServerSetting::unique_key_bitmap_cache_size_bytes];
    double unique_key_bitmap_cache_size_ratio = server_settings[ServerSetting::unique_key_bitmap_cache_size_ratio];
    if (unique_key_bitmap_cache_size > max_cache_size)
    {
        unique_key_bitmap_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered UNIQUE KEY delete-bitmap cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(unique_key_bitmap_cache_size));
    }
    global_context->setDeleteBitmapCache(unique_key_bitmap_cache_policy_name, unique_key_bitmap_cache_size, unique_key_bitmap_cache_size_ratio);

    String primary_index_cache_policy = server_settings[ServerSetting::primary_index_cache_policy];
    size_t primary_index_cache_size = server_settings[ServerSetting::primary_index_cache_size];
    double primary_index_cache_size_ratio = server_settings[ServerSetting::primary_index_cache_size_ratio];
    if (primary_index_cache_size > max_cache_size)
    {
        primary_index_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered primary index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(primary_index_cache_size));
    }
    global_context->setPrimaryIndexCache(primary_index_cache_policy, primary_index_cache_size, primary_index_cache_size_ratio);

    String statistics_cache_policy = server_settings[ServerSetting::statistics_cache_policy];
    size_t statistics_cache_size = server_settings[ServerSetting::statistics_cache_size];
    double statistics_cache_size_ratio = server_settings[ServerSetting::statistics_cache_size_ratio];
    if (statistics_cache_size > max_cache_size)
    {
        statistics_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered statistics cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(statistics_cache_size));
    }
    global_context->setStatisticsCache(statistics_cache_policy, statistics_cache_size, statistics_cache_size_ratio);
    String columns_cache_policy = server_settings[ServerSetting::columns_cache_policy];
    /// Unless configured explicitly, the columns cache is sized relative to the memory of the server.
    size_t columns_cache_size = config().getUInt64("columns_cache_size",
        getDefaultColumnsCacheSize(physical_server_memory, server_settings[ServerSetting::columns_cache_size_to_ram_ratio]));
    double columns_cache_size_ratio = server_settings[ServerSetting::columns_cache_size_ratio];
    if (columns_cache_size > max_cache_size)
    {
        columns_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered columns cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(columns_cache_size));
    }
    global_context->setColumnsCache(columns_cache_policy, columns_cache_size, columns_cache_size_ratio);
    /// The columns cache gives its memory back to the queries when the server is short of it, see
    /// `ColumnsCache::autoResize`: on every tick of the memory worker, and when an allocation is
    /// about to exceed the memory limit.
    if (auto columns_cache = global_context->getColumnsCache())
    {
        columns_cache->setAutoResizeSettings(
            server_settings[ServerSetting::columns_cache_free_memory_ratio],
            server_settings[ServerSetting::columns_cache_history_window_ms]);
        setMemoryReleasableCache(columns_cache.get());
        memory_worker.setReleasableCache(columns_cache);
    }

    String index_uncompressed_cache_policy = server_settings[ServerSetting::index_uncompressed_cache_policy];
    size_t index_uncompressed_cache_size = server_settings[ServerSetting::index_uncompressed_cache_size];
    double index_uncompressed_cache_size_ratio = server_settings[ServerSetting::index_uncompressed_cache_size_ratio];
    if (index_uncompressed_cache_size > max_cache_size)
    {
        index_uncompressed_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered index uncompressed cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(index_uncompressed_cache_size));
    }
    global_context->setIndexUncompressedCache(index_uncompressed_cache_policy, index_uncompressed_cache_size, index_uncompressed_cache_size_ratio);

    String index_mark_cache_policy = server_settings[ServerSetting::index_mark_cache_policy];
    size_t index_mark_cache_size = server_settings[ServerSetting::index_mark_cache_size];
    double index_mark_cache_size_ratio = server_settings[ServerSetting::index_mark_cache_size_ratio];
    if (index_mark_cache_size > max_cache_size)
    {
        index_mark_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered index mark cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(index_mark_cache_size));
    }
    global_context->setIndexMarkCache(index_mark_cache_policy, index_mark_cache_size, index_mark_cache_size_ratio);

    String vector_similarity_index_cache_policy = server_settings[ServerSetting::vector_similarity_index_cache_policy];
    size_t vector_similarity_index_cache_size = server_settings[ServerSetting::vector_similarity_index_cache_size];
    size_t vector_similarity_index_cache_max_entries = server_settings[ServerSetting::vector_similarity_index_cache_max_entries];
    double vector_similarity_index_cache_size_ratio = server_settings[ServerSetting::vector_similarity_index_cache_size_ratio];
    if (vector_similarity_index_cache_size > max_cache_size)
    {
        vector_similarity_index_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered vector similarity index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(vector_similarity_index_cache_size));
    }
    global_context->setVectorSimilarityIndexCache(vector_similarity_index_cache_policy, vector_similarity_index_cache_size, vector_similarity_index_cache_max_entries, vector_similarity_index_cache_size_ratio);

    String text_index_tokens_cache_policy = server_settings[ServerSetting::text_index_tokens_cache_policy];
    size_t text_index_tokens_cache_size = server_settings[ServerSetting::text_index_tokens_cache_size];
    size_t text_index_tokens_cache_max_count = server_settings[ServerSetting::text_index_tokens_cache_max_entries];
    double text_index_tokens_cache_size_ratio = server_settings[ServerSetting::text_index_tokens_cache_size_ratio];
    if (text_index_tokens_cache_size > max_cache_size)
    {
        text_index_tokens_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered text index tokens cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(text_index_tokens_cache_size));
    }
    global_context->setTextIndexTokensCache(text_index_tokens_cache_policy, text_index_tokens_cache_size, text_index_tokens_cache_max_count, text_index_tokens_cache_size_ratio);

    String text_index_header_cache_policy = server_settings[ServerSetting::text_index_header_cache_policy];
    size_t text_index_header_cache_size = server_settings[ServerSetting::text_index_header_cache_size];
    size_t text_index_header_cache_max_count = server_settings[ServerSetting::text_index_header_cache_max_entries];
    double text_index_header_cache_size_ratio = server_settings[ServerSetting::text_index_header_cache_size_ratio];
    if (text_index_header_cache_size > max_cache_size)
    {
        text_index_header_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered text index header cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(text_index_header_cache_size));
    }
    global_context->setTextIndexHeaderCache(text_index_header_cache_policy, text_index_header_cache_size, text_index_header_cache_max_count, text_index_header_cache_size_ratio);

    String text_index_postings_cache_policy = server_settings[ServerSetting::text_index_postings_cache_policy];
    size_t text_index_postings_cache_size = server_settings[ServerSetting::text_index_postings_cache_size];
    size_t text_index_postings_cache_max_count = server_settings[ServerSetting::text_index_postings_cache_max_entries];
    double text_index_postings_cache_size_ratio = server_settings[ServerSetting::text_index_postings_cache_size_ratio];
    if (text_index_postings_cache_size > max_cache_size)
    {
        text_index_postings_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered text index posting list cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(text_index_postings_cache_size));
    }
    global_context->setTextIndexPostingsCache(text_index_postings_cache_policy, text_index_postings_cache_size, text_index_postings_cache_max_count, text_index_postings_cache_size_ratio);

    size_t mmap_cache_size = server_settings[ServerSetting::mmap_cache_size];
    if (mmap_cache_size > max_cache_size)
    {
        mmap_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered mmap file cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(mmap_cache_size));
    }
    global_context->setMMappedFileCache(mmap_cache_size);

#if USE_AVRO
    String iceberg_metadata_files_cache_policy = server_settings[ServerSetting::iceberg_metadata_files_cache_policy];
    size_t iceberg_metadata_files_cache_size = server_settings[ServerSetting::iceberg_metadata_files_cache_size];
    size_t iceberg_metadata_files_cache_max_entries = server_settings[ServerSetting::iceberg_metadata_files_cache_max_entries];
    double iceberg_metadata_files_cache_size_ratio = server_settings[ServerSetting::iceberg_metadata_files_cache_size_ratio];
    if (iceberg_metadata_files_cache_size > max_cache_size)
    {
        iceberg_metadata_files_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered Iceberg metadata cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(iceberg_metadata_files_cache_size));
    }
    global_context->setIcebergMetadataFilesCache(iceberg_metadata_files_cache_policy, iceberg_metadata_files_cache_size, iceberg_metadata_files_cache_max_entries, iceberg_metadata_files_cache_size_ratio);

    String paimon_metadata_files_cache_policy = server_settings[ServerSetting::paimon_metadata_files_cache_policy];
    size_t paimon_metadata_files_cache_size = server_settings[ServerSetting::paimon_metadata_files_cache_size];
    size_t paimon_metadata_files_cache_max_entries = server_settings[ServerSetting::paimon_metadata_files_cache_max_entries];
    double paimon_metadata_files_cache_size_ratio = server_settings[ServerSetting::paimon_metadata_files_cache_size_ratio];
    if (paimon_metadata_files_cache_size > max_cache_size)
    {
        paimon_metadata_files_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered Paimon metadata cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(paimon_metadata_files_cache_size));
    }
    global_context->setPaimonMetadataFilesCache(paimon_metadata_files_cache_policy, paimon_metadata_files_cache_size, paimon_metadata_files_cache_max_entries, paimon_metadata_files_cache_size_ratio);
#endif
#if USE_PARQUET
    String parquet_metadata_cache_policy = server_settings[ServerSetting::parquet_metadata_cache_policy];
    size_t parquet_metadata_cache_size = server_settings[ServerSetting::parquet_metadata_cache_size];
    size_t parquet_metadata_cache_max_entries = server_settings[ServerSetting::parquet_metadata_cache_max_entries];
    double parquet_metadata_cache_size_ratio = server_settings[ServerSetting::parquet_metadata_cache_size_ratio];
    if (parquet_metadata_cache_size > max_cache_size)
    {
        parquet_metadata_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered Parquet metadata cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(parquet_metadata_cache_size));
    }
    global_context->setParquetMetadataCache(parquet_metadata_cache_policy, parquet_metadata_cache_size, parquet_metadata_cache_max_entries, parquet_metadata_cache_size_ratio);
#endif

    Names allowed_disks_table_engines;
    splitInto<','>(allowed_disks_table_engines, server_settings[ServerSetting::allowed_disks_for_table_engines].value);
    global_context->setAllowedDisksForTableEngines(std::unordered_set<String>(allowed_disks_table_engines.begin(), allowed_disks_table_engines.end()));

    String query_condition_cache_policy = server_settings[ServerSetting::query_condition_cache_policy];
    size_t query_condition_cache_size = server_settings[ServerSetting::query_condition_cache_size];
    double query_condition_cache_size_ratio = server_settings[ServerSetting::query_condition_cache_size_ratio];
    if (query_condition_cache_size > max_cache_size)
    {
        query_condition_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered query condition cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(query_condition_cache_size));
    }
    global_context->setQueryConditionCache(query_condition_cache_policy, query_condition_cache_size, query_condition_cache_size_ratio);

    String encryption_header_cache_policy = server_settings[ServerSetting::encryption_header_cache_policy];
    size_t encryption_header_cache_size = server_settings[ServerSetting::encryption_header_cache_size];
    double encryption_header_cache_size_ratio = server_settings[ServerSetting::encryption_header_cache_size_ratio];
    if (encryption_header_cache_size > max_cache_size)
    {
        encryption_header_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered encryption header cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(encryption_header_cache_size));
    }
    global_context->setEncryptionHeaderCache(encryption_header_cache_policy, encryption_header_cache_size, encryption_header_cache_size_ratio);

    size_t query_result_cache_max_size_in_bytes = server_settings[ServerSetting::query_cache_max_size_in_bytes];
    size_t query_result_cache_max_entries = server_settings[ServerSetting::query_cache_max_entries];
    size_t query_result_cache_max_entry_size_in_bytes = server_settings[ServerSetting::query_cache_max_entry_size_in_bytes];
    size_t query_result_cache_max_entry_size_in_rows = server_settings[ServerSetting::query_cache_max_entry_size_in_rows];
    if (query_result_cache_max_size_in_bytes > max_cache_size)
    {
        query_result_cache_max_size_in_bytes = max_cache_size;
        LOG_INFO(log, "Lowered query result cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(query_result_cache_max_size_in_bytes));
    }
    global_context->setQueryResultCache(query_result_cache_max_size_in_bytes, query_result_cache_max_entries, query_result_cache_max_entry_size_in_bytes, query_result_cache_max_entry_size_in_rows);

#if USE_EMBEDDED_COMPILER
    size_t compiled_expression_cache_max_size_in_bytes = server_settings[ServerSetting::compiled_expression_cache_size];
    size_t compiled_expression_cache_max_elements = server_settings[ServerSetting::compiled_expression_cache_elements_size];
    CompiledExpressionCacheFactory::instance().init(compiled_expression_cache_max_size_in_bytes, compiled_expression_cache_max_elements);
#endif

    size_t point_in_polygon_cache_size = server_settings[ServerSetting::point_in_polygon_cache_size];
    if (point_in_polygon_cache_size > max_cache_size)
    {
        point_in_polygon_cache_size = max_cache_size;
        LOG_INFO(log, "Lowered point in polygon cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(point_in_polygon_cache_size));
    }
    setPointInPolygonCacheMaxSizeInBytes(point_in_polygon_cache_size);

    NamedCollectionFactory::instance().loadIfNot();
    SQLDefinedHandlersFactory::instance().loadIfNot();
    FileCacheFactory::instance().loadDefaultCaches(config(), global_context);

    /// Initialize main config reloader.
    std::string include_from_path = server_settings[ServerSetting::include_from];

    if (config().has("query_masking_rules"))
    {
        SensitiveDataMasker::setInstance(std::make_unique<SensitiveDataMasker>(config(), "query_masking_rules"));
    }

    std::optional<CgroupsMemoryUsageObserver> cgroups_memory_usage_observer;
    try
    {
        const auto & wait_time = server_settings[ServerSetting::cgroups_memory_usage_observer_wait_time];
        if (wait_time != 0)
            cgroups_memory_usage_observer.emplace(std::chrono::seconds(wait_time));
    }
    catch (Exception &)
    {
        tryLogCurrentException(log, "Disabling cgroup memory observer because of an error during initialization");
    }

    /// TLS certificates, keys and CA certificates are reloaded by CertificateReloader when these files change.
    std::vector<std::string> extra_paths = {include_from_path};
    auto watch_path = [&](const std::string & file_path)
    {
        if (!file_path.empty())
            extra_paths.emplace_back(file_path);
    };
    watch_path(server_settings[ServerSetting::openssl_server_certificate_file]);
    watch_path(server_settings[ServerSetting::openssl_server_private_key_file]);
    watch_path(server_settings[ServerSetting::openssl_server_ca_config]);
    watch_path(server_settings[ServerSetting::openssl_client_certificate_file]);
    watch_path(server_settings[ServerSetting::openssl_client_private_key_file]);
    watch_path(server_settings[ServerSetting::openssl_client_ca_config]);

    Poco::Util::AbstractConfiguration::Keys protocols;
    config().keys("protocols", protocols);
    for (const auto & protocol : protocols)
        for (const auto * key : {"certificateFile", "privateKeyFile", "caConfig"})
            watch_path(config().getString("protocols." + protocol + "." + key, ""));

    DNSResolver::instance().setFilterSettings(server_settings[ServerSetting::dns_allow_resolve_names_to_ipv4], server_settings[ServerSetting::dns_allow_resolve_names_to_ipv6]);
    /// DNSCacheUpdater uses BackgroundSchedulePool which lives in shared context
    /// and thus this object must be created after the SCOPE_EXIT object where shared
    /// context is destroyed.
    /// In addition this object has to be created before any scenario that requires
    /// DNS resolution (e.g. the loading of tables and clusters config).
    std::unique_ptr<DNSCacheUpdater> dns_cache_updater;
    if (server_settings[ServerSetting::disable_internal_dns_cache])
    {
        /// Disable DNS caching at all
        DNSResolver::instance().setDisableCacheFlag();
        LOG_DEBUG(log, "DNS caching disabled");
    }
    else
    {
        DNSResolver::instance().setCacheMaxEntries(server_settings[ServerSetting::dns_cache_max_entries]);

        /// Prime the local host name / addresses cache before the updater starts,
        /// so that the first DNSCacheUpdater tick on a healthy server does not
        /// see an empty cache and spuriously report "IPs of host name X have been changed".
        DNSResolver::instance().updateHostNameAndAddresses();

        /// Initialize a watcher periodically updating DNS cache
        dns_cache_updater = std::make_unique<DNSCacheUpdater>(
            global_context, server_settings[ServerSetting::dns_cache_update_period], server_settings[ServerSetting::dns_max_consecutive_failures]);
        dns_cache_updater->start();
    }

    /// Whether `--skip_check_for_incorrect_settings=1` was passed on the *command line*, captured
    /// independently of any config file layer. Reusing the layered `config()` here would conflate
    /// the command-line value with whatever the *previously loaded* file set: a config that once
    /// contained `<skip_check_for_incorrect_settings>1</skip_check_for_incorrect_settings>` would
    /// leave that flag live in `config()`, so a reload that *removes* the flag while adding an
    /// unknown key would still be skipped — diverging from a fresh startup with the same file,
    /// which would reject it. Re-parsing `argv()` with `argsToConfig` (exactly as `BaseDaemon`
    /// populates the command-line layer) reproduces the command-line-only view; the priority is
    /// irrelevant because the throwaway config has a single layer.
    bool command_line_skip_check = false;
    {
        Poco::AutoPtr<Poco::Util::LayeredConfiguration> command_line_config(new Poco::Util::LayeredConfiguration);
        argsToConfig(argv(), *command_line_config, PRIO_DEFAULT);
        command_line_skip_check = command_line_config->getBool("skip_check_for_incorrect_settings", false);
    }

    auto main_config_reloader = std::make_unique<ConfigReloader>(
        config_path,
        extra_paths,
        server_settings[ServerSetting::path],
        std::move(main_config_zk_node_cache),
        main_config_zk_changed_event,
        [&](ConfigurationPtr loaded_config, bool initial_loading)
        {
            /// Validate the new config BEFORE installing it into the global layered config, so a
            /// failed reload does not leave `config()` partially mutated with unvalidated changes.
            /// The escape hatch is honored when it is either passed on the command line (persisted
            /// across reloads) or present in the *new* file being validated — never inherited from
            /// the previously loaded file layer (see `command_line_skip_check` above). This matches
            /// startup: validating the same file fresh must reach the same accept/reject decision.
            const bool skip_check
                = command_line_skip_check || loaded_config->getBool("skip_check_for_incorrect_settings", false);
            if (!skip_check)
                Settings::checkNoSettingNamesAtTopLevel(*loaded_config, config_path);
            /// Same as on initial load: validate the reloaded XML config rather than the layered
            /// view, so CLI-injected and Poco-internal top-level keys do not need an allowlist.
            ServerSettings::checkUnknownSettings(*loaded_config, config_path, skip_check);

            /// Fail closed on a legacy insert_deduplication_version, or on a memory-pressure threshold
            /// triple that is out of range or out of order, arriving via a runtime reload.
            /// Validate the incoming config BEFORE config().replace below: validating after would mutate
            /// the live config even for a rejected reload (ConfigReloader has no rollback hook), leaving
            /// system.server_settings reporting an unsupported value. Reject first, then replace.
            {
                ServerSettings incoming_server_settings;
                incoming_server_settings.loadSettingsFromConfig(*loaded_config);
                validate_insert_deduplication_version(incoming_server_settings);
                validateMemoryPressureThresholds(
                    incoming_server_settings[ServerSetting::reader_executor_memory_pressure_elevated_level_pct],
                    incoming_server_settings[ServerSetting::reader_executor_memory_pressure_high_level_pct],
                    incoming_server_settings[ServerSetting::reader_executor_memory_pressure_critical_level_pct]);
            }

            /// The Prometheus check below, the reloaders and `latest_config` keep the configuration they are given, so it must have
            /// the command line on top of the file and, like `config()`, resolve `${application.*}` and `${system.*}` references.
            Poco::AutoPtr<Poco::Util::LayeredConfiguration> file_and_command_line(new Poco::Util::LayeredConfiguration);
            file_and_command_line->add(loaded_config, PRIO_DEFAULT);
            argsToConfig(argv(), *file_and_command_line, PRIO_APPLICATION - 100);
            file_and_command_line->add(application_properties, PRIO_APPLICATION, /* shared */ true);
            file_and_command_line->add(new Poco::Util::SystemConfiguration, PRIO_SYSTEM, /* shared */ false);
            const ConfigurationPtr reload_config = file_and_command_line;

            /// Fail closed on a Prometheus constant label that collides with a label an endpoint writes
            /// itself. `asynchronous_metrics_key_values_mode` takes part in this check, because it decides
            /// whether the key of a key-value asynchronous metric is written as a label (`device="sda"`),
            /// so the same set of constant labels can be unambiguous under one form and not under another.
            /// Validate the incoming config BEFORE config().replace below, for the same reason as above:
            /// the form is read from the live configuration on every update of the asynchronous metrics,
            /// so validating afterwards would leave a rejected reload publishing the new form while the
            /// endpoints keep serving with the labels of the old one.
            validatePrometheusConstantLabels(
                *reload_config, servedHTTPHandlersKeys(*reload_config), hasPrometheusListener(*reload_config));

            config().replace("default", loaded_config, PRIO_DEFAULT, true);

            ServerSettings new_server_settings;
            new_server_settings.loadSettingsFromConfig(config());

            /// The check above sees the incoming file alone. A triple split across the file and the
            /// command-line layer can have each source valid and the merge out of order, so check the
            /// merged view too - here, before the first live setting below is touched.
            validateMemoryPressureThresholds(
                new_server_settings[ServerSetting::reader_executor_memory_pressure_elevated_level_pct],
                new_server_settings[ServerSetting::reader_executor_memory_pressure_high_level_pct],
                new_server_settings[ServerSetting::reader_executor_memory_pressure_critical_level_pct]);

            DB::abort_on_logical_error.store(new_server_settings[ServerSetting::abort_on_logical_error], std::memory_order_relaxed);

            ConnectionRegistry::instance().setEnabled(new_server_settings[ServerSetting::collect_connection_metrics]);

            /// The seccomp filter cannot be changed once installed; `system.server_settings` keeps showing the mode in force.
            if (const auto requested_seccomp_mode = getRequestedSeccompMode();
                requested_seccomp_mode && *requested_seccomp_mode != new_server_settings[ServerSetting::seccomp].value)
                LOG_WARNING(
                    log,
                    "The `seccomp` server setting was changed from `{}` to `{}` in the configuration, but it takes effect only "
                    "after a restart: the seccomp policy of a running process cannot be changed",
                    SettingFieldSeccompMode(*requested_seccomp_mode).toString(),
                    new_server_settings[ServerSetting::seccomp].toString());

            size_t max_server_memory_usage = new_server_settings[ServerSetting::max_server_memory_usage];
            const double max_server_memory_usage_to_ram_ratio = new_server_settings[ServerSetting::max_server_memory_usage_to_ram_ratio];
            const size_t current_physical_server_memory = getMemoryAmount(); /// With cgroups, the amount of memory available to the server can be changed dynamically.
            const size_t default_max_server_memory_usage = static_cast<size_t>(static_cast<double>(current_physical_server_memory) * max_server_memory_usage_to_ram_ratio);

            if (max_server_memory_usage == 0)
            {
                max_server_memory_usage = default_max_server_memory_usage;
                LOG_INFO(log, "Changed setting 'max_server_memory_usage' to {}"
                    " ({} available memory * {:.2f} max_server_memory_usage_to_ram_ratio)",
                    formatReadableSizeWithBinarySuffix(max_server_memory_usage),
                    formatReadableSizeWithBinarySuffix(current_physical_server_memory),
                    max_server_memory_usage_to_ram_ratio);
            }
            else if (max_server_memory_usage > default_max_server_memory_usage)
            {
                max_server_memory_usage = default_max_server_memory_usage;
                LOG_INFO(log, "Lowered setting 'max_server_memory_usage' to {}"
                    " because the system has too little memory. The new value was"
                    " calculated as {} available memory * {:.2f} max_server_memory_usage_to_ram_ratio",
                    formatReadableSizeWithBinarySuffix(max_server_memory_usage),
                    formatReadableSizeWithBinarySuffix(current_physical_server_memory),
                    max_server_memory_usage_to_ram_ratio);
            }

            total_memory_tracker.setDescription("(total)");
            total_memory_tracker.setMetric(CurrentMetrics::MemoryTracking);

            /// Inform `MemoryWorker` of the configured ceiling and the ratio so its dynamic
            /// adjustment (which only sees `MemAvailable` or cgroup memory) cannot exceed
            /// the explicit `max_server_memory_usage`, and so a config-reload change to
            /// `max_server_memory_usage_to_ram_ratio` takes effect on the next worker tick.
            /// `setDynamicHardLimitSettings` also installs `max_server_memory_usage` as the
            /// new hard limit atomically with the settings update; doing it outside that call
            /// would re-open a reload race against the worker tick.
            /// A zero ratio disables the runtime adjustment, keeping only the static cap;
            /// `memory_worker_dynamic_hard_limit = 0` requests exactly that.
            memory_worker.setDynamicHardLimitSettings(
                static_cast<Int64>(max_server_memory_usage),
                new_server_settings[ServerSetting::memory_worker_dynamic_hard_limit]
                    ? max_server_memory_usage_to_ram_ratio
                    : 0.0);

            CurrentMemoryTracker::setMinAllocationSizeBytesToThrow(
                new_server_settings[ServerSetting::min_allocation_size_to_throw_on_memory_limit]);

            MemoryTracker::setMinAllocationSizeToLogStackTrace(
                new_server_settings[ServerSetting::min_allocation_size_to_log_stack_trace]);

            per_cpu_memory.setBudgetCapacity(new_server_settings[ServerSetting::max_per_cpu_untracked_memory]);
            per_cpu_memory.setThreadBuffer(new_server_settings[ServerSetting::per_cpu_untracked_memory_thread_buffer]);

            size_t merges_mutations_memory_usage_soft_limit = new_server_settings[ServerSetting::merges_mutations_memory_usage_soft_limit];

            const double merges_mutations_memory_usage_to_ram_ratio = new_server_settings[ServerSetting::merges_mutations_memory_usage_to_ram_ratio];
            size_t default_merges_mutations_server_memory_usage = static_cast<size_t>(static_cast<double>(current_physical_server_memory) * merges_mutations_memory_usage_to_ram_ratio);
            if (merges_mutations_memory_usage_soft_limit == 0)
            {
                merges_mutations_memory_usage_soft_limit = default_merges_mutations_server_memory_usage;
                LOG_INFO(log, "Setting merges_mutations_memory_usage_soft_limit was set to {}"
                    " ({} available * {:.2f} merges_mutations_memory_usage_to_ram_ratio)",
                    formatReadableSizeWithBinarySuffix(merges_mutations_memory_usage_soft_limit),
                    formatReadableSizeWithBinarySuffix(current_physical_server_memory),
                    merges_mutations_memory_usage_to_ram_ratio);
            }
            else if (merges_mutations_memory_usage_soft_limit > default_merges_mutations_server_memory_usage)
            {
                merges_mutations_memory_usage_soft_limit = default_merges_mutations_server_memory_usage;
                LOG_WARNING(log, "Setting merges_mutations_memory_usage_soft_limit was set to {}"
                    " ({} available * {:.2f} merges_mutations_memory_usage_to_ram_ratio)",
                    formatReadableSizeWithBinarySuffix(merges_mutations_memory_usage_soft_limit),
                    formatReadableSizeWithBinarySuffix(current_physical_server_memory),
                    merges_mutations_memory_usage_to_ram_ratio);
            }

            LOG_INFO(log, "Merges and mutations memory limit is set to {}",
                formatReadableSizeWithBinarySuffix(merges_mutations_memory_usage_soft_limit));
            background_memory_tracker.setSoftLimit(merges_mutations_memory_usage_soft_limit);
            background_memory_tracker.setDescription("(background)");
            background_memory_tracker.setMetric(CurrentMetrics::MergesMutationsMemoryTracking);

            auto * global_overcommit_tracker = global_context->getGlobalOvercommitTracker();
            total_memory_tracker.setOvercommitTracker(global_overcommit_tracker);

            // FIXME logging-related things need synchronization -- see the 'Logger * log' saved
            // in a lot of places. For now, disable updating log configuration without server restart.
            //setTextLog(global_context->getTextLog());
            updateLevels(config(), logger());
            /// File-only, so `remote_servers` ignores the command line: `Cluster` takes the server's own `tcp_port` from this configuration
            /// to find its local replicas, and with `-- --tcp_port` the replicas on the file's port would become remote.
            global_context->setClustersConfig(loaded_config, has_zookeeper);
            global_context->setMacros(std::make_unique<Macros>(config(), "macros", log));
            global_context->setExternalAuthenticatorsConfig(config());

            global_context->setDashboardsConfig(config());

            if (global_context->isServerCompletelyStarted())
            {
                /// It does not make sense to reload anything before server has started.
                /// Moreover, it may break initialization order.
                global_context->loadOrReloadDictionaries(config());
                global_context->loadUserDefinedExecutableFunctionDrivers(config());
                global_context->loadOrReloadUserDefinedExecutableFunctions(config());
            }

            global_context->setRemoteHostFilter(config());
            global_context->setHTTPHeaderFilter(config());

            global_context->setMaxTableSizeToDrop(new_server_settings[ServerSetting::max_table_size_to_drop]);
            global_context->setMaxPartitionSizeToDrop(new_server_settings[ServerSetting::max_partition_size_to_drop]);
            global_context->setMaxNamedCollectionNumToWarn(new_server_settings[ServerSetting::max_named_collection_num_to_warn]);
            global_context->setMaxTableNumToWarn(new_server_settings[ServerSetting::max_table_num_to_warn]);
            global_context->setMaxViewNumToWarn(new_server_settings[ServerSetting::max_view_num_to_warn]);
            global_context->setMaxDictionaryNumToWarn(new_server_settings[ServerSetting::max_dictionary_num_to_warn]);
            global_context->setMaxDatabaseNumToWarn(new_server_settings[ServerSetting::max_database_num_to_warn]);
            global_context->setMaxPartNumToWarn(new_server_settings[ServerSetting::max_part_num_to_warn]);
            global_context->setMaxNamedCollectionNumToThrow(new_server_settings[ServerSetting::max_named_collection_num_to_throw]);
            global_context->setMaxTableNumToThrow(new_server_settings[ServerSetting::max_table_num_to_throw]);
            global_context->setMaxReplicatedTableNumToThrow(new_server_settings[ServerSetting::max_replicated_table_num_to_throw]);
            global_context->setMaxViewNumToThrow(new_server_settings[ServerSetting::max_view_num_to_throw]);
            global_context->setMaxDictionaryNumToThrow(new_server_settings[ServerSetting::max_dictionary_num_to_throw]);
            global_context->setMaxDatabaseNumToThrow(new_server_settings[ServerSetting::max_database_num_to_throw]);
            global_context->setMaxPendingMutationsToWarn(new_server_settings[ServerSetting::max_pending_mutations_to_warn]);
            global_context->setMaxPendingMutationsExecutionTimeToWarn(new_server_settings[ServerSetting::max_pending_mutations_execution_time_to_warn]);
            global_context->getAccessControl().setAllowTierSettings(new_server_settings[ServerSetting::allow_feature_tier]);
            global_context->setUsersToIgnoreEarlyMemoryLimitCheck(new_server_settings[ServerSetting::users_to_ignore_early_memory_limit_check]);
            global_context->allowSystemAllocateMemory(config().getBool("allow_system_allocate_memory", false));

            global_context->setS3QueueDisableStreaming(new_server_settings[ServerSetting::s3queue_disable_streaming]);
            global_context->setReadThroughDistributedCache(new_server_settings[ServerSetting::enable_read_through_distributed_cache]);
            global_context->setWriteThroughDistributedCache(new_server_settings[ServerSetting::enable_write_through_distributed_cache]);

            global_context->setOSCPUOverloadSettings(static_cast<double>(new_server_settings[ServerSetting::min_os_cpu_wait_time_ratio_to_drop_connection]), static_cast<double>(new_server_settings[ServerSetting::max_os_cpu_wait_time_ratio_to_drop_connection]));

            size_t remote_read_bandwidth = new_server_settings[ServerSetting::max_remote_read_network_bandwidth_for_server];
            size_t remote_write_bandwidth = new_server_settings[ServerSetting::max_remote_write_network_bandwidth_for_server];

            global_context->reloadRemoteThrottlerConfig(remote_read_bandwidth,remote_write_bandwidth);
            LOG_INFO(log, "Setting max_remote_read_network_bandwidth_for_server was set to {}", remote_read_bandwidth);
            LOG_INFO(log, "Setting max_remote_write_network_bandwidth_for_server was set to {}", remote_write_bandwidth);

            size_t local_read_bandwidth = new_server_settings[ServerSetting::max_local_read_bandwidth_for_server];
            size_t local_write_bandwidth = new_server_settings[ServerSetting::max_local_write_bandwidth_for_server];

            global_context->reloadLocalThrottlerConfig(local_read_bandwidth,local_write_bandwidth);
            LOG_INFO(log, "Setting max_local_read_bandwidth_for_server was set to {}", local_read_bandwidth);
            LOG_INFO(log, "Setting max_local_write_bandwidth_for_server was set to {}", local_write_bandwidth);

            size_t max_remote_read_connections = new_server_settings[ServerSetting::max_remote_read_connections];
            global_context->reloadLongConnectionLimitConfig(max_remote_read_connections);
            LOG_INFO(log, "Setting max_remote_read_connections was set to {}", max_remote_read_connections);

#if ENABLE_DISTRIBUTED_CACHE
            for (const auto & distr_cache_instance : distr_cache_instances)
                distr_cache_instance->updateConfig(config());
#endif

            /// Only for system.server_settings
            global_context->setConfigReloaderInterval(new_server_settings[ServerSetting::config_reload_interval_ms]);

            auto [concurrent_threads_soft_limit, concurrency_control_scheduler] = global_context->setConcurrentThreadsSoftLimit(
                new_server_settings[ServerSetting::concurrent_threads_soft_limit_num],
                new_server_settings[ServerSetting::concurrent_threads_soft_limit_ratio_to_cores],
                new_server_settings[ServerSetting::concurrent_threads_scheduler],
                new_server_settings[ServerSetting::concurrent_threads_lazy_allocation]);
            LOG_INFO(log, "ConcurrencyControl limit is set to {} CPU slots with '{}' scheduler",
                concurrent_threads_soft_limit == UnlimitedSlots ? std::string("UNLIMITED") : std::to_string(concurrent_threads_soft_limit),
                concurrency_control_scheduler);

            global_context->getProcessList().setMaxSize(new_server_settings[ServerSetting::max_concurrent_queries]);
            global_context->getProcessList().setMaxInsertQueriesAmount(new_server_settings[ServerSetting::max_concurrent_insert_queries]);
            global_context->getProcessList().setMaxSelectQueriesAmount(new_server_settings[ServerSetting::max_concurrent_select_queries]);
            global_context->getProcessList().setMaxWaitingQueriesAmount(new_server_settings[ServerSetting::max_waiting_queries]);

            if (config().has("keeper_server"))
            {
                global_context->updateKeeperConfiguration(config());
            }

            /// Reload the number of threads for global pools.
            /// Note: If you specified it in the top level config (not it config of default profile)
            /// then ClickHouse will use it exactly.
            /// This is done for backward compatibility.
            if (global_context->areBackgroundExecutorsInitialized())
            {
                const auto & new_pool_size = new_server_settings[ServerSetting::background_pool_size];
                const auto & new_ratio = new_server_settings[ServerSetting::background_merges_mutations_concurrency_ratio];
                global_context->getMergeMutateExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, static_cast<size_t>(static_cast<double>(new_pool_size) * static_cast<double>(new_ratio)));
                global_context->getMergeMutateExecutor()->updateSchedulingPolicy(new_server_settings[ServerSetting::background_merges_mutations_scheduling_policy].toString());
            }

            if (global_context->areBackgroundExecutorsInitialized())
            {
                const auto & new_pool_size = new_server_settings[ServerSetting::background_move_pool_size];
                global_context->getMovesExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, new_pool_size);
            }

            if (global_context->areBackgroundExecutorsInitialized())
            {
                const auto & new_pool_size = new_server_settings[ServerSetting::background_fetches_pool_size];
                global_context->getFetchesExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, new_pool_size);
            }

            if (global_context->areBackgroundExecutorsInitialized())
            {
                const auto & new_pool_size = new_server_settings[ServerSetting::background_common_pool_size];
                global_context->getCommonExecutor()->increaseThreadsAndMaxTasksCount(new_pool_size, new_pool_size);
            }

            global_context->getBufferFlushSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_buffer_flush_schedule_pool_size]);
            global_context->getSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_schedule_pool_size]);
            global_context->getMessageBrokerSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_message_broker_schedule_pool_size]);
            global_context->getDistributedSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_distributed_schedule_pool_size]);
            global_context->getStreamingSchedulePool()->increaseThreadsCount(new_server_settings[ServerSetting::background_streaming_schedule_pool_size]);

            global_context->getAsyncLoader().setMaxThreads(TablesLoaderForegroundPoolId, new_server_settings[ServerSetting::tables_loader_foreground_pool_size]);
            global_context->getAsyncLoader().setMaxThreads(TablesLoaderBackgroundLoadPoolId, new_server_settings[ServerSetting::tables_loader_background_pool_size]);
            global_context->getAsyncLoader().setMaxThreads(TablesLoaderBackgroundStartupPoolId, new_server_settings[ServerSetting::tables_loader_background_pool_size]);

            getIOThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_io_thread_pool_size],
                new_server_settings[ServerSetting::max_io_thread_pool_free_size],
                new_server_settings[ServerSetting::io_thread_pool_queue_size]);

            getBackupsIOThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_backups_io_thread_pool_size],
                new_server_settings[ServerSetting::max_backups_io_thread_pool_free_size],
                new_server_settings[ServerSetting::backups_io_thread_pool_queue_size]);

            getFetchPartitionThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_fetch_partition_thread_pool_size],
                0, // FETCH PARTITION is relatively rare, no need to keep threads
                new_server_settings[ServerSetting::max_fetch_partition_thread_pool_size]);

            getActivePartsLoadingThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_active_parts_loading_thread_pool_size],
                0, // We don't need any threads once all the parts will be loaded
                new_server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]);

            getOutdatedPartsLoadingThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size],
                0, // We don't need any threads once all the parts will be loaded
                new_server_settings[ServerSetting::max_outdated_parts_loading_thread_pool_size]);

            /// It could grow if we need to synchronously wait until all the data parts will be loaded.
            getOutdatedPartsLoadingThreadPool().setMaxTurboThreads(
                new_server_settings[ServerSetting::max_active_parts_loading_thread_pool_size]
            );

            getPartsCleaningThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_parts_cleaning_thread_pool_size],
                0, // We don't need any threads one all the parts will be deleted
                new_server_settings[ServerSetting::max_parts_cleaning_thread_pool_size]);

            getMergeTreePrefixesDeserializationThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_size],
                new_server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_free_size],
                new_server_settings[ServerSetting::prefixes_deserialization_thread_pool_thread_pool_queue_size]);

            getFormatParsingThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::max_format_parsing_thread_pool_size],
                new_server_settings[ServerSetting::max_format_parsing_thread_pool_free_size],
                new_server_settings[ServerSetting::format_parsing_thread_pool_queue_size]);

            getAIRequestThreadPool().reloadConfiguration(
                new_server_settings[ServerSetting::ai_function_max_thread_pool_size],
                new_server_settings[ServerSetting::ai_function_max_thread_pool_free_size],
                new_server_settings[ServerSetting::ai_function_thread_pool_queue_size]);

            global_context->setMergeWorkload(new_server_settings[ServerSetting::merge_workload]);
            global_context->setMutationWorkload(new_server_settings[ServerSetting::mutation_workload]);
            global_context->setThrowOnUnknownWorkload(new_server_settings[ServerSetting::throw_on_unknown_workload]);
            global_context->setCPUSlotPreemption(
                new_server_settings[ServerSetting::cpu_slot_preemption],
                new_server_settings[ServerSetting::cpu_slot_quantum_ns],
                new_server_settings[ServerSetting::cpu_slot_preemption_timeout_ms]);
            global_context->setCPUSlotParking(new_server_settings[ServerSetting::cpu_slot_parking]);

            /// The thresholds were validated above, so a rejected reload never reaches here. Sets and stamps
            /// the shared thresholds, so every monitor adopts them on its next sample,
            /// bypassing the sticky cooldown.
            setMemoryPressureThresholds(
                new_server_settings[ServerSetting::reader_executor_memory_pressure_elevated_level_pct],
                new_server_settings[ServerSetting::reader_executor_memory_pressure_high_level_pct],
                new_server_settings[ServerSetting::reader_executor_memory_pressure_critical_level_pct]);

            if (config().has("resources") || config().has("workload_classifiers"))
            {
                LOG_WARNING(
                    &logger(),
                    "Config-based resource scheduling ('resources' and 'workload_classifiers' configuration sections) "
                    "has been removed and is ignored. Use 'CREATE RESOURCE' and 'CREATE WORKLOAD' queries instead.");
            }

            /// Load WORKLOADs and RESOURCEs.
            global_context->getWorkloadEntityStoragePtr()->loadEntities(config());

            if (!initial_loading)
            {
                /// We do not load ZooKeeper configuration on the first config loading
                /// because TestKeeper server is not started yet.
                if (zkutil::hasZooKeeperConfig(*reload_config))
                    global_context->reloadZooKeeperIfChanged(reload_config);

                global_context->reloadAuxiliaryZooKeepersConfigIfChanged(reload_config);
                global_context->reloadQueryMaskingRulesIfChanged(reload_config);

                if (global_context->isServerCompletelyStarted())
                {
                    std::lock_guard lock(servers_lock);
                    updateServers(config(), new_server_settings, server_pool, *async_metrics, servers, servers_to_start_before_tables);
                }
            }

            global_context->updateStorageConfiguration(config());
            global_context->updateInterserverCredentials(config());

            {
                const size_t max_cache_size_in_bytes = static_cast<size_t>(
                    static_cast<double>(current_physical_server_memory) * new_server_settings[ServerSetting::cache_size_to_ram_max_ratio]);

                global_context->updateUncompressedCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateColumnsCacheConfiguration(
                    config(),
                    getDefaultColumnsCacheSize(current_physical_server_memory, new_server_settings[ServerSetting::columns_cache_size_to_ram_ratio]),
                    max_cache_size_in_bytes);
                if (auto columns_cache = global_context->getColumnsCache())
                    columns_cache->setAutoResizeSettings(
                        new_server_settings[ServerSetting::columns_cache_free_memory_ratio],
                        new_server_settings[ServerSetting::columns_cache_history_window_ms]);
                global_context->updateMarkCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateUniqueKeyIndexCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateDeleteBitmapCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updatePrimaryIndexCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateStatisticsCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateIndexUncompressedCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateIndexMarkCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateVectorSimilarityIndexCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateTextIndexTokensCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateTextIndexHeaderCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateTextIndexPostingsCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateMMappedFileCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateQueryResultCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateQueryConditionCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updateEncryptionHeaderCacheConfiguration(config(), max_cache_size_in_bytes);
                setPointInPolygonCacheMaxSizeInBytes(
                    std::min<size_t>(new_server_settings[ServerSetting::point_in_polygon_cache_size], max_cache_size_in_bytes));
#if USE_AVRO
                global_context->updateIcebergMetadataFilesCacheConfiguration(config(), max_cache_size_in_bytes);
                global_context->updatePaimonMetadataFilesCacheConfiguration(config(), max_cache_size_in_bytes);
#endif
#if USE_PARQUET
                global_context->updateParquetMetadataCacheConfiguration(config(), max_cache_size_in_bytes);
#endif
            }

#if USE_SSL
            CertificateReloader::instance().tryReloadAll(config());
#endif
            CompressionCodecEncrypted::Configuration::instance().tryLoad(config(), "encryption_codecs");
            global_context->reloadStatelessWorkersConfig(config());
            NamedCollectionFactory::instance().reloadFromConfig(config());
            FileCacheFactory::instance().updateSettingsFromConfig(config());

            HTTPConnectionPools::instance().setLimits(
                HTTPConnectionPools::Limits{
                    new_server_settings[ServerSetting::disk_connections_soft_limit],
                    new_server_settings[ServerSetting::disk_connections_warn_limit],
                    new_server_settings[ServerSetting::disk_connections_store_limit],
                    new_server_settings[ServerSetting::disk_connections_hard_limit],
                },
                HTTPConnectionPools::Limits{
                    new_server_settings[ServerSetting::storage_connections_soft_limit],
                    new_server_settings[ServerSetting::storage_connections_warn_limit],
                    new_server_settings[ServerSetting::storage_connections_store_limit],
                    new_server_settings[ServerSetting::storage_connections_hard_limit],
                },
                HTTPConnectionPools::Limits{
                    new_server_settings[ServerSetting::http_connections_soft_limit],
                    new_server_settings[ServerSetting::http_connections_warn_limit],
                    new_server_settings[ServerSetting::http_connections_store_limit],
                    new_server_settings[ServerSetting::http_connections_hard_limit],
                });

            HTTPConnectionPools::instance().setSocketBufferSizes(
                HTTPConnectionPools::SocketBufferSizes{
                    new_server_settings[ServerSetting::disk_connections_rcvbuf],
                    new_server_settings[ServerSetting::disk_connections_sndbuf],
                },
                HTTPConnectionPools::SocketBufferSizes{
                    new_server_settings[ServerSetting::storage_connections_rcvbuf],
                    new_server_settings[ServerSetting::storage_connections_sndbuf],
                },
                HTTPConnectionPools::SocketBufferSizes{
                    new_server_settings[ServerSetting::http_connections_rcvbuf],
                    new_server_settings[ServerSetting::http_connections_sndbuf],
                });

            DNSResolver::instance().setFilterSettings(new_server_settings[ServerSetting::dns_allow_resolve_names_to_ipv4], new_server_settings[ServerSetting::dns_allow_resolve_names_to_ipv6]);

            if (global_context->isServerCompletelyStarted())
                CannotAllocateThreadFaultInjector::setFaultProbability(new_server_settings[ServerSetting::cannot_allocate_thread_fault_injection_probability]);

            /// Update core dump size limit.
            {
                rlimit rlim{};
                if (getrlimit(RLIMIT_CORE, &rlim) == 0)
                {
                    rlim.rlim_cur = config().getUInt64("core_dump.size_limit", 1024 * 1024 * 1024);
                    if (setrlimit(RLIMIT_CORE, &rlim))
                        LOG_WARNING(log, "Cannot set max size of core file to {}", rlim.rlim_cur);
                }
            }

            ProfileEvents::increment(ProfileEvents::MainConfigLoads);

            /// Must be the last.
            latest_config = reload_config;
        });

    const auto listen_hosts = getListenHosts(config());
    const auto interserver_listen_hosts = getInterserverListenHosts(config());
    const auto listen_try = getListenTry(config(), server_settings);

    if (config().has("keeper_server.server_id"))
    {
#if USE_NURAFT
        //// If we don't have configured connection probably someone trying to use clickhouse-server instead
        //// of clickhouse-keeper, so start synchronously.
        bool can_initialize_keeper_async = false;

        if (has_zookeeper) /// We have configured connection to some zookeeper cluster
        {
            /// If we cannot connect to some other node from our cluster then we have to wait our Keeper start
            /// synchronously.
            can_initialize_keeper_async = global_context->tryCheckClientConnectionToMyKeeperCluster();
        }
        /// Initialize certificates before Keeper RAFT so that NuRaft SSL contexts
        /// can register with CertificateReloader for hot-reload support.
#if USE_SSL
        CertificateReloader::instance().tryLoad(config());
        CertificateReloader::instance().tryLoadClient(config());
#endif

        /// Initialize keeper RAFT.
        global_context->initializeKeeperDispatcher(can_initialize_keeper_async);
        FourLetterCommandFactory::registerCommands(*global_context->getKeeperDispatcher());

        auto config_getter = [this] () -> const Poco::Util::AbstractConfiguration &
        {
            return global_context->getConfigRef();
        };

        for (const auto & listen_host : listen_hosts)
        {
            /// TCP Keeper
            const char * port_name = "keeper_server.tcp_port";
            createServer(
                config(), listen_host, port_name, listen_try, /* start_server: */ false,
                servers_to_start_before_tables,
                [&](UInt16 port) -> ProtocolServerAdapter
                {
                    Poco::Net::ServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port);
                    socket.setReceiveTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_receive_timeout_sec], 0));
                    socket.setSendTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_send_timeout_sec], 0));
                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::END,
                        "Keeper (tcp): " + address.toString(),
                        std::make_unique<TCPServer>(
                            new KeeperTCPHandlerFactory(
                                config_getter,
                                global_context->getKeeperDispatcher(),
                                global_context->getSettingsRef()[Setting::receive_timeout].totalSeconds(),
                                global_context->getSettingsRef()[Setting::send_timeout].totalSeconds(),
                                false),
                            server_pool,
                            socket));
                });

            const char * secure_port_name = "keeper_server.tcp_port_secure";
            createServer(
                config(), listen_host, secure_port_name, listen_try, /* start_server: */ false,
                servers_to_start_before_tables,
                [&](UInt16 port) -> ProtocolServerAdapter
                {
#if USE_SSL
                    Poco::Net::SecureServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                    socket.setReceiveTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_receive_timeout_sec], 0));
                    socket.setSendTimeout(Poco::Timespan(server_settings[ServerSetting::keeper_server_socket_send_timeout_sec], 0));
                    return ProtocolServerAdapter(
                        listen_host,
                        secure_port_name,
                        ServerType::Type::END,
                        "Keeper with secure protocol (tcp_secure): " + address.toString(),
                        std::make_unique<TCPServer>(
                            new KeeperTCPHandlerFactory(
                                config_getter,
                                global_context->getKeeperDispatcher(),
                                global_context->getSettingsRef()[Setting::receive_timeout].totalSeconds(),
                                global_context->getSettingsRef()[Setting::send_timeout].totalSeconds(),
                                true),
                            server_pool,
                            socket));
#else
                    UNUSED(port);
                    throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support.");
#endif
                });

            /// HTTP control endpoints
            port_name = "keeper_server.http_control.port";
            if (!config().getString(port_name, "").empty())
            {
                auto handler_factory = createKeeperHTTPHandlerFactory(
                    *this, config_getter(), global_context->getKeeperDispatcher(), "KeeperHTTPHandler-factory");
                createServer(config(), listen_host, port_name, listen_try, /* start_server: */ false,
                servers_to_start_before_tables,
                [&](UInt16 port) -> ProtocolServerAdapter
                {
                    auto http_context = httpContext();
                    Poco::Timespan keep_alive_timeout(server_settings[ServerSetting::keep_alive_timeout].totalSeconds(), 0);
                    Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams;
                    http_params->setTimeout(http_context->getReceiveTimeout());
                    http_params->setKeepAliveTimeout(keep_alive_timeout);

                    Poco::Net::ServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port);
                    socket.setReceiveTimeout(http_context->getReceiveTimeout());
                    socket.setSendTimeout(http_context->getSendTimeout());
                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::END,
                        "HTTP Control: http://" + address.toString(),
                        std::make_unique<HTTPServer>(
                            std::move(http_context),
                            handler_factory,
                            server_pool,
                            socket,
                            http_params));
                });
            }

            /// HTTPS control endpoints
            port_name = "keeper_server.http_control.secure_port";
            if (!config().getString(port_name, "").empty())
            {
#if USE_SSL
                auto handler_factory = createKeeperHTTPHandlerFactory(
                    *this, config_getter(), global_context->getKeeperDispatcher(), "KeeperHTTPSHandler-factory");
#endif
                createServer(config(), listen_host, port_name, listen_try, /* start_server: */ false,
                servers_to_start_before_tables,
                [&](UInt16 port) -> ProtocolServerAdapter
                {
#if USE_SSL
                    auto http_context = httpContext();
                    Poco::Timespan keep_alive_timeout(server_settings[ServerSetting::keep_alive_timeout].totalSeconds(), 0);
                    Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams;
                    http_params->setTimeout(http_context->getReceiveTimeout());
                    http_params->setKeepAliveTimeout(keep_alive_timeout);

                    Poco::Net::SecureServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                    socket.setReceiveTimeout(http_context->getReceiveTimeout());
                    socket.setSendTimeout(http_context->getSendTimeout());
                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::END,
                        "HTTPS Control: https://" + address.toString(),
                        std::make_unique<HTTPServer>(
                            std::move(http_context),
                            handler_factory,
                            server_pool,
                            socket,
                            http_params));
#else
                    UNUSED(port);
                    throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "HTTPS control protocol is disabled because Poco library was built without NetSSL support.");
#endif
                });
            }
        }
#else
        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "ClickHouse server built without NuRaft library. Cannot use internal coordination.");
#endif

    }

    {
        std::lock_guard lock(servers_lock);
        /// We should start interserver communications before (and more important shutdown after) tables.
        /// Because server can wait for a long-running queries (for example in tcp_handler) after interserver handler was already shut down.
        /// In this case we will have replicated tables which are unable to send any parts to other replicas, but still can
        /// communicate with zookeeper, execute merges, etc.
        createInterserverServers(
            config(),
            server_settings,
            interserver_listen_hosts,
            listen_try,
            server_pool,
            *async_metrics,
            servers_to_start_before_tables,
            /* start_servers= */ false);
    }

#if USE_SSL
    /// We may notice that try to reload certificates twice within this function.
    /// First time here before starting the `servers_to_start_before_tables`, and then
    /// one more time before starting the `servers`. The problem we're trying to avoid is:
    /// 1. We create a socket for keeper secure protocol (w/o starting a server yet). That creates a SSL context (in particular, it might be the default context).
    /// 2. We then start this server a few lines below. While handling the first request, it will access (read) certs.
    /// 3. A little further down, we reload certificates before starting the rest of the servers.
    /// 4. `CertificateReloader` will set its custom `cert_cb` to the default context if it is not initialized yet (see `CertificateReloader::init()`).
    /// 5. Items (2) and (4) are not synchronized, so there might be a data race for example (see https://github.com/ClickHouse/ClickHouse/issues/85412).
    CertificateReloader::instance().tryLoad(config());
    CertificateReloader::instance().tryLoadClient(config());
#endif

    {
        std::lock_guard lock(servers_lock);
        startServers(servers_to_start_before_tables, listen_try, log);
        for (const auto & server : servers_to_start_before_tables)
        {
            if (server.bindsOnStart())
                global_context->registerServerPort(server.getPortName(), static_cast<UInt16>(config().getInt(server.getPortName())));
        }
    }

#if USE_SSL
    /// ACME client is necessary for CertificateReloader to work in case Let's Encrypt is configured,
    /// but can not start before Keeper server can be initialized (in case of embedded Keeper).
    /// Let's try deferring until servers are started.
    ACME::Client::instance().initialize(config());
#endif

    const auto & cache_disk_name = server_settings[ServerSetting::temporary_data_in_cache].value;

    /// Storage with temporary data for processing of heavy queries.
    if (!server_settings[ServerSetting::tmp_policy].value.empty())
    {
        global_context->setTemporaryStoragePolicy(
            server_settings[ServerSetting::tmp_policy], server_settings[ServerSetting::max_temporary_data_on_disk_size]);
    }
    else if (!cache_disk_name.empty())
    {
        /// When cache is used as a temporary data storage with load_metadata_asynchronously
        /// the cache might not be fully ready when setTemporaryStorageInCache() is called
        /// below which might try to use the cache before it's fully loaded leading to
        /// `File cache is not initialized` errors. Instead disallow load_metadata_asynchronously
        /// for cache disks used as temporary storage.
        auto cache_name = global_context->getDisk(cache_disk_name)->getCacheName();
        if (!cache_name.empty())
        {
            auto cache_data = FileCacheFactory::instance().getByName(cache_name);
            if (cache_data->getSettings()[FileCacheSetting::load_metadata_asynchronously])
            {
                throw Exception(
                    ErrorCodes::INVALID_SETTING_VALUE,
                    "Cache disk '{}' is used as a temporary storage, load_metadata_asynchronously is disallowed.",
                    cache_name);
            }
        }
        global_context->setTemporaryStorageInCache(
            server_settings[ServerSetting::temporary_data_in_cache], server_settings[ServerSetting::max_temporary_data_on_disk_size]);
    }
    else
    {
        const auto & tmp_path_setting = server_settings[ServerSetting::tmp_path];
        std::string temporary_path = tmp_path_setting.changed
            ? getCanonicalPath(String(tmp_path_setting.value), path_str) : String(path / "tmp/");
        global_context->setTemporaryStoragePath(temporary_path, server_settings[ServerSetting::max_temporary_data_on_disk_size]);
    }

    /// Initialize access storages.
    auto & access_control = global_context->getAccessControl();
    try
    {
        access_control.setupFromMainConfig(config(), config_path, [&] { return global_context->getZooKeeper(); });
    }
    catch (...)
    {
        tryLogCurrentException(log, "Caught exception while setting up access control.");
        throw;
    }

    auto disable_config_reload_callbacks = [&]
    {
        global_context->setConfigReloadCallback([]
        {
            throw Exception(ErrorCodes::ABORTED, "Cannot reload config because the server is shutting down");
        });
        if (cgroups_memory_usage_observer)
            cgroups_memory_usage_observer->setOnMemoryAmountAvailableChangedFn([] {});
    };
    SCOPE_EXIT({ disable_config_reload_callbacks(); });

    if (cgroups_memory_usage_observer)
    {
        cgroups_memory_usage_observer->setOnMemoryAmountAvailableChangedFn([&]() { main_config_reloader->reload(); });
        cgroups_memory_usage_observer->startThread();
    }

    /// Reload config in SYSTEM RELOAD CONFIG query.
    global_context->setConfigReloadCallback([&]()
    {
        main_config_reloader->reload();
        access_control.reload(AccessControl::ReloadMode::USERS_CONFIG_ONLY);
    });

    global_context->setStopServersCallback([&](const ServerType & server_type)
    {
        std::lock_guard lock(servers_lock);
        stopServers(servers, server_type);
    });

    global_context->setStartServersCallback([&](const ServerType & server_type)
    {
        std::lock_guard lock(servers_lock);
        /// Introspection server can run SYSTEM START LISTEN before other servers' startup or after other servers' shutdown.
        /// In either case, we need to return an error.
        if (is_cancelled || !global_context->isServerCompletelyStarted())
            throw Exception(ErrorCodes::ABORTED, "Cannot start listeners because the server is starting up or shutting down");
        /// The config may have been reloaded since startup, so recompute the listener
        /// configuration from the current config instead of the startup snapshot.
        ServerSettings current_server_settings;
        current_server_settings.loadSettingsFromConfig(config());
        createServers(
            config(),
            current_server_settings,
            getListenHosts(config()),
            getListenTry(config(), current_server_settings),
            server_pool,
            *async_metrics,
            servers,
            /* start_servers= */ true,
            server_type);
    });

    /// Limit on total number of concurrently executed queries.
    global_context->getProcessList().setMaxSize(server_settings[ServerSetting::max_concurrent_queries]);

    /// Load global settings from default_profile and system_profile.
    global_context->setDefaultProfiles(config());

    /// Initialize background executors after we load default_profile config.
    /// This is needed to load proper values of background_pool_size etc.
    global_context->initializeBackgroundExecutorsIfNeeded();

    if (server_settings[ServerSetting::async_insert_threads])
    {
        global_context->setAsynchronousInsertQueue(std::make_shared<AsynchronousInsertQueue>(
            global_context,
            server_settings[ServerSetting::async_insert_threads],
            server_settings[ServerSetting::async_insert_queue_flush_on_shutdown]));
    }

    /// Set path for format schema files
    const auto & format_schema_path_setting = server_settings[ServerSetting::format_schema_path];
    fs::path format_schema_path(format_schema_path_setting.changed
        ? fs::path(getCanonicalPath(String(format_schema_path_setting.value), path_str)) : path / "format_schemas/");
    global_context->setFormatSchemaPath(format_schema_path);
    fs::create_directories(format_schema_path);

    /// Set the path for google proto files
    if (server_settings[ServerSetting::google_protos_path].changed)
        global_context->setGoogleProtosPath(fs::weakly_canonical(server_settings[ServerSetting::google_protos_path].value));

    /// Set path for filesystem caches
    {
        String filesystem_caches_path = server_settings[ServerSetting::filesystem_caches_path].value;
        if (!filesystem_caches_path.empty())
            global_context->setFilesystemCachesPath(getCanonicalPath(std::move(filesystem_caches_path), path_str));
    }

    /// NOTE: Do sanity checks after we loaded all possible substitutions (for the configuration) from ZK
    /// Additionally, making the check after the default profile is initialized.
    /// It is important to initialize MergeTreeSettings after Settings, to support compatibility for MergeTreeSettings.
    addEnvironmentWarnings(global_context, logger(), path_str, server_settings[ServerSetting::logger_log]);

    /// Check sanity of MergeTreeSettings on server startup
    {
        size_t background_pool_tasks = global_context->getMergeMutateExecutor()->getMaxTasksCount();
        global_context->getMergeTreeSettings().sanityCheck(background_pool_tasks, global_context->wasBackgroundPoolAutoLowered());
        global_context->getReplicatedMergeTreeSettings().sanityCheck(
            background_pool_tasks, global_context->wasBackgroundPoolAutoLowered());
    }
    /// try set up encryption. There are some errors in config, error will be printed and server wouldn't start.
    CompressionCodecEncrypted::Configuration::instance().load(config(), "encryption_codecs");

    auto replicas_reconnector = ReplicasReconnector::init(global_context);

    /// Set current database name before loading tables and databases because
    /// system logs may copy global context.
    std::string default_database = server_settings[ServerSetting::default_database];
    if (default_database.empty())
        throw Exception(ErrorCodes::BAD_ARGUMENTS, "default_database cannot be empty");
    global_context->setCurrentDatabaseNameInGlobalContext(default_database);

    /// Start collecting asynchronous metrics before loading tables, so that the Prometheus endpoint
    /// (started below) exposes meaningful values during the potentially long metadata loading phase.
    /// This includes the OS/jemalloc metrics and the per-table metrics such as
    /// `TotalIndexGranularityBytesInMemoryAllocated`, which let one observe memory growth as tables
    /// are loaded. The metric computation skips not-yet-loaded tables (just like with
    /// `async_load_databases`), and writing to `system.asynchronous_metric_log` is skipped until the
    /// system logs are initialized below. The asynchronous metrics thread reads the `servers` lists
    /// under `servers_lock`, so it is safe to start before the main `servers` exist.
    async_metrics->start();
    global_context->setAsynchronousMetrics(async_metrics.get());

    /// Start the Prometheus endpoint before loading tables, so that metrics stay observable during
    /// the potentially long metadata loading phase. Only do it for metrics-only configurations:
    /// custom `prometheus.handlers` may serve queries (`remote_write`, `remote_read`, `query`,
    /// `api_v1`) and must follow the regular `servers` lifecycle. The server is created in the
    /// regular `servers` list, so runtime reconfiguration and `SYSTEM START/STOP LISTEN` handle it
    /// as usual. The later `createServers` call skips it (`createServer` does not recreate a live
    /// server), and the start loop for `servers` skips it too (`ProtocolServerAdapter::start` does
    /// nothing for an already started server).
    if (!config().has("prometheus.handlers"))
    {
        std::lock_guard lock(servers_lock);
        createServers(
            config(),
            server_settings,
            listen_hosts,
            listen_try,
            server_pool,
            *async_metrics,
            servers,
            /* start_servers= */ true,
            ServerType(ServerType::Type::PROMETHEUS));
    }

    LOG_INFO(log, "Loading metadata from {}", path_str);

    LoadTaskPtrs load_system_metadata_tasks;
    LoadTaskPtrs load_metadata_tasks;

    // Make sure that if exception is thrown during startup async, new async loading jobs are not going to be called.
    // This is important for the case when exception is thrown from loading of metadata with `async_load_databases = false`
    // to avoid simultaneously running table startups and destructing databases.
    SCOPE_EXIT_SAFE(
        LOG_INFO(log, "Stopping AsyncLoader.");

        // Waits for all currently running jobs to finish and do not run any other pending jobs.
        global_context->getAsyncLoader().shutdown();
    );

    /// The introspection servers accept TCP connections while the server is otherwise unreachable:
    /// before attaching and after detaching of tables.
    if (hasIntrospectionProtocols(config()))
    {
        global_context->setStopIntrospectionServersCallback([&]
        {
            if (introspection_servers.empty())
                return;

            LOG_DEBUG(log, "Waiting for current connections to introspection servers to finish.");
            size_t current_connections = 0;
            {
                std::lock_guard lock(servers_lock);
                for (auto & server : introspection_servers)
                {
                    server.stop();
                    current_connections += server.currentConnections();
                }
            }

            if (current_connections)
                LOG_INFO(log, "Closed all introspection listening sockets. Waiting for {} outstanding connections.", current_connections);
            else
                LOG_INFO(log, "Closed all introspection listening sockets.");

            global_context->getProcessList().killAllQueries();

            if (current_connections)
                current_connections = waitServersToFinish(
                    introspection_servers, servers_lock, server_settings[ServerSetting::shutdown_wait_unfinished]);

            if (current_connections)
            {
                dumpCoverageReportIfPossible();
                LOG_WARNING(log, "Closed connections to introspection servers. But {} remain. Will shutdown forcefully.", current_connections);
                /// No leak check: remaining connection handlers still own memory it would report as
                /// leaked. Reported, because here stderr is the server log.
                safeExit(0, LeakCheck::SkipAndReport);
            }

            LOG_INFO(log, "Closed connections to introspection servers.");
            introspection_server_pool->joinAll();
        });

        introspection_server_pool = std::make_unique<Poco::ThreadPool>(
            /* minCapacity */ 1,
            /* maxCapacity */ server_settings[ServerSetting::max_connections],
            /* idleTime */ 60,
            /* stackSize */ DEFAULT_THREAD_STACK_SIZE ? static_cast<int>(DEFAULT_THREAD_STACK_SIZE) : POCO_THREAD_STACK_SIZE,
            server_settings[ServerSetting::global_profiler_real_time_period_ns],
            server_settings[ServerSetting::global_profiler_cpu_time_period_ns]);

        std::lock_guard lock(servers_lock);
        createServers(
            config(),
            server_settings,
            listen_hosts,
            listen_try,
            *introspection_server_pool,
            *async_metrics,
            introspection_servers,
            /* start_servers= */ true,
            ServerType(ServerType::Type::QUERIES_ALL),
            /* only_introspection_protocols= */ true);
    }

    try
    {
        /// Don't run background queries until we loaded tables.
        /// (In particular things would break if a background drop query happens before the
        /// loadMarkedAsDroppedTables() call below - it'll see dropped table metadata and try to
        /// drop the table a second time and throw an exception.)
        global_context->getRefreshSet().setRefreshesStopped(true);

        auto & database_catalog = DatabaseCatalog::instance();
        /// We load temporary database first, because projections need it.
        database_catalog.initializeAndLoadTemporaryDatabase();
        /// The tables (including the ones in the `system` database) can depend on the dictionaries from the configuration files,
        /// so register them before loading the databases. They start loading later, in `loadOrReloadDictionaries`.
        global_context->registerDictionariesBeforeLoadingMetadata(config());
        load_system_metadata_tasks = loadMetadataSystem(global_context, server_settings[ServerSetting::async_load_system_database]);
        maybeConvertSystemDatabase(global_context, load_system_metadata_tasks);

        /// Startup scripts can depend on the system log tables.
        if (config().has("startup_scripts") && !server_settings[ServerSetting::prepare_system_log_tables_on_startup].changed)
            global_context->setServerSetting("prepare_system_log_tables_on_startup", true);

        /// After attaching system databases we can initialize system log.
        global_context->initializeSystemLogs();

        if (has_trace_collector)
            global_context->initializeTraceCollector();

#if defined(OS_LINUX)
        if (oom_canary)
            oom_canary->start();
#endif

        global_context->handleSystemZooKeeperConnectionLogAfterInitializationIfNeeded();

        /// Build loggers before tables startup to make log messages from tables
        /// attach available in system.text_log
        buildLoggers(config(), logger());
        initializeAzureSDKLogger(server_settings, logger().getLevel());
        /// After the system database is created, attach virtual system tables (in addition to query_log and part_log)
        bool has_keeper_server = false;
#if USE_NURAFT
        has_keeper_server = global_context->tryGetKeeperDispatcher() != nullptr;
#endif
        attachSystemTablesServer(global_context, *database_catalog.getSystemDatabase(), has_zookeeper, has_keeper_server);
        attachInformationSchema(global_context, *database_catalog.getDatabase(DatabaseCatalog::INFORMATION_SCHEMA));
        attachInformationSchema(global_context, *database_catalog.getDatabase(DatabaseCatalog::INFORMATION_SCHEMA_UPPERCASE));
        /// Firstly remove partially dropped databases, to avoid race with Materialized...SyncThread,
        /// that may execute DROP before loadMarkedAsDroppedTables() in background,
        /// and so loadMarkedAsDroppedTables() will find it and try to add, and UUID will overlap.
        database_catalog.loadMarkedAsDroppedTables();
        database_catalog.createBackgroundTasks();
        /// Then, load remaining databases (some of them maybe be loaded asynchronously)
        load_metadata_tasks = loadMetadata(global_context, default_database, server_settings[ServerSetting::async_load_databases]);
        /// If we need to convert database engines, disable async tables loading
        convertDatabasesEnginesIfNeed(load_metadata_tasks, global_context);
        database_catalog.startupBackgroundTasks();
        /// After loading validate that default database exists
        database_catalog.assertDatabaseExists(default_database);
        /// Load user-defined SQL functions.
        global_context->loadUserDefinedExecutableFunctionDrivers(config());
        global_context->getUserDefinedSQLObjectsStorage().loadObjects();

        /// For driver-based executable UDFs persisted as ATTACH FUNCTION queries, ensure the
        /// dynamic configuration files exist; re-run their drivers if they are missing.
        UserDefinedSQLFunctionFactory::instance().reloadDriverBasedFunctions(
            global_context, global_context->getUserDefinedSQLObjectsStorage());

        global_context->getRefreshSet().setRefreshesStopped(false);
    }
    catch (...)
    {
        tryLogCurrentException(log, "Caught exception while loading metadata");
        throw;
    }

    DatabaseCatalog::instance().startReplicatedDDLQueries();

    LOG_DEBUG(log, "Loaded metadata.");

#if defined(OS_LINUX)
    auto tasks_stats_provider = TasksStatsCounters::findBestAvailableProvider();
    if (tasks_stats_provider == TasksStatsCounters::MetricsProvider::None)
    {
        LOG_INFO(log, "It looks like this system does not have procfs mounted at /proc location."
            " 'taskstats' performance statistics will be disabled."
            " It could happen due to incorrect ClickHouse package installation.");
    }
    else
    {
        LOG_INFO(log, "Tasks stats provider: {}", TasksStatsCounters::metricsProviderString(tasks_stats_provider));
    }

    if (!hasLinuxCapability(CAP_SYS_NICE))
    {
        LOG_INFO(log, "It looks like the process has no CAP_SYS_NICE capability, the setting 'os_thread_priority' will have no effect."
            " It could happen due to incorrect ClickHouse package installation."
            " You could resolve the problem manually with 'sudo setcap cap_sys_nice=+ep {}'."
            " Note that it will not work on 'nosuid' mounted filesystems.",
            executable_path);
    }
#else
    LOG_INFO(log, "TaskStats is not implemented for this OS. IO accounting will be disabled.");
#endif

    {
        attachSystemTablesAsync(global_context, *DatabaseCatalog::instance().getSystemDatabase(), *async_metrics);

        {
            std::lock_guard lock(servers_lock);
            createServers(config(), server_settings, listen_hosts, listen_try, server_pool, *async_metrics, servers);
            if (servers.empty())
                throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG,
                                "No servers started (add valid listen_host and 'tcp_port' or 'http_port' "
                                "to configuration file.)");
        }

        if (servers.empty())
             throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG,
                             "No servers started (add valid listen_host and 'tcp_port' or 'http_port' "
                             "to configuration file.)");

#if USE_SSL
        ACME::Client::instance().initialize(config());
        CertificateReloader::instance().tryLoad(config());
        CertificateReloader::instance().tryLoadClient(config());
#endif

        main_config_reloader->start();
        access_control.startPeriodicReloading();

        /// try to load dictionaries immediately, throw on error and die
        try
        {
            global_context->loadOrReloadDictionaries(config());

            if (!server_settings[ServerSetting::dictionaries_lazy_load] && server_settings[ServerSetting::wait_dictionaries_load_at_startup])
                global_context->waitForDictionariesLoad();
        }
        catch (...)
        {
            tryLogCurrentException(log, "Caught exception while loading dictionaries.");
            throw;
        }

        /// try to load embedded dictionaries immediately, throw on error and die
        try
        {
            global_context->tryCreateEmbeddedDictionaries();
        }
        catch (...)
        {
            tryLogCurrentException(log, "Caught exception while loading embedded dictionaries.");
            throw;
        }

        /// try to load user defined executable functions, throw on error and die
        try
        {
            global_context->loadOrReloadUserDefinedExecutableFunctions(config());
        }
        catch (...)
        {
            tryLogCurrentException(log, "Caught exception while loading user defined executable functions.");
            throw;
        }

        if (has_zookeeper && config().has("distributed_ddl"))
        {
            /// DDL worker should be started after all tables were loaded
            String ddl_queue_path = server_settings[ServerSetting::distributed_ddl_path];
            String ddl_replicas_path = server_settings[ServerSetting::distributed_ddl_replicas_path];
            int pool_size = server_settings[ServerSetting::distributed_ddl_pool_size];
            if (pool_size < 1)
                throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "distributed_ddl.pool_size should be greater then 0");
            global_context->setDDLWorker(
                std::make_unique<DDLWorker>(
                    pool_size,
                    ddl_queue_path,
                    ddl_replicas_path,
                    global_context,
                    &config(),
                    "distributed_ddl",
                    "default",
                    "DDLWorker",
                    &CurrentMetrics::MaxDDLEntryID,
                    &CurrentMetrics::MaxPushedDDLEntryID),
                joinTasks(load_system_metadata_tasks, load_metadata_tasks));
        }

        /// Do not keep tasks in server, they should be kept inside databases. Used here to make dependent tasks only.
        load_system_metadata_tasks.clear();
        load_system_metadata_tasks.shrink_to_fit();
        load_metadata_tasks.clear();
        load_metadata_tasks.shrink_to_fit();

        if (config().has("startup_scripts"))
            loadStartupScripts(config(), server_settings, global_context, log);

        auto stop_acme_instance = []{
#if USE_SSL
            /// Stop ACME tasks.
            ACME::Client::instance().shutdown();
#endif
        };

        /// Wrapping the call to OOM canary stop in a lambda lets us write
        /// the OS_LINUX guard outside the SCOPE_EXIT_SAFE macro argument list,
        /// avoiding -Wembedded-directive.
        auto stop_oom_canary = [&]{
#if defined(OS_LINUX)
            if (oom_canary)
                oom_canary->stop();
#endif
        };

        SCOPE_EXIT_SAFE({
            const auto & logger_shutdown_level_setting = server_settings[ServerSetting::logger_shutdown_level];
            if (logger_shutdown_level_setting.changed && !logger_shutdown_level_setting.value.empty())
            {
                /// Set the root logger level to the shutdown level.
                /// This is useful for debugging shutdown issues.
                config().setString("logger.level", logger_shutdown_level_setting.value);
                Loggers::updateLevels(config(), logger());

                LOG_INFO(log, "Set root logger in level {} before shutdown", logger_shutdown_level_setting.value);
            }

            if (config().has("logger.shutdown_console_log_level") && !config().getString("logger.shutdown_console_log_level").empty())
            {
                /// Set the root logger level to the shutdown level.
                /// This is useful for debugging shutdown issues.
                config().setString("logger.console_log_level", config().getString("logger.shutdown_console_log_level"));
                Loggers::updateLevels(config(), logger());

                LOG_INFO(log, "Set console logger in level {} before shutdown", config().getString("logger.shutdown_console_log_level"));
            }

            LOG_DEBUG(log, "Received termination signal.");

            CurrentMetrics::set(CurrentMetrics::IsServerShuttingDown, 1);

            /// Stop reloading of the main config. This must be done before everything else because it
            /// can try to access/modify already deleted objects.
            /// E.g. it can recreate new servers or it may pass a changed config to some destroyed parts of ContextSharedPart.
            disable_config_reload_callbacks();
            main_config_reloader.reset();
            access_control.stopPeriodicReloading();

            stop_acme_instance();

            is_cancelled = true;

            LOG_DEBUG(log, "Waiting for current connections to close.");

            size_t current_connections = 0;
            {
                std::lock_guard lock(servers_lock);
                for (auto & server : servers)
                {
                    server.stop();
                    current_connections += server.currentConnections();
                }
            }

            global_context->getBackgroundQueryPool().finish();
            global_context->getRefreshSet().setRefreshesStopped(true);

            if (current_connections)
                LOG_WARNING(log, "Closed all listening sockets. Waiting for {} outstanding connections.", current_connections);
            else
                LOG_INFO(log, "Closed all listening sockets.");

            /// Wait for unfinished backups and restores.
            /// This must be done after closing listening sockets (no more socket-delivered backups/restores) but before ProcessList::killAllQueries
            /// (because killAllQueries() will cancel all running backups/restores).
            bool backups_finished = true;
            if (server_settings[ServerSetting::shutdown_wait_backups_and_restores])
                global_context->waitAllBackupsAndRestores();
            else
            {
                /// Refused first so that the wait cannot miss an operation started after it took
                /// its snapshot; a distributed DDL query can still deliver one here.
                global_context->stopAcceptingNewBackupsAndRestores();
                backups_finished = global_context->cancelAllBackupsAndRestores(
                    std::chrono::steady_clock::now()
                    + std::chrono::seconds(server_settings[ServerSetting::shutdown_wait_unfinished]));
            }

            stop_oom_canary();

            /// Killing remaining queries.
            /// In wait mode the DDL worker's parts of ON CLUSTER backups and restores are spared, and waited for below.
            /// Any other backup or restore that missed the wait above is cancelled: an initiator's own part needs this host's DDL worker.
            const bool wait_backups_and_restores = server_settings[ServerSetting::shutdown_wait_backups_and_restores];
            if (!server_settings[ServerSetting::shutdown_wait_unfinished_queries])
                global_context->getProcessList().killAllQueries(/* except_ddl_backups_and_restores= */ wait_backups_and_restores);

            size_t wait_limit_seconds = server_settings[ServerSetting::shutdown_wait_unfinished];
            auto wait_start = std::chrono::steady_clock::now();
            auto shutdown_deadline = wait_start + std::chrono::milliseconds(wait_limit_seconds * 1000);

            if (current_connections)
                current_connections = waitServersToFinish(servers, servers_lock, wait_limit_seconds);

            if (current_connections)
                LOG_WARNING(log, "Closed connections. But {} remain."
                    " Tip: To increase wait time add to config: <shutdown_wait_unfinished>300</shutdown_wait_unfinished>", current_connections);
            else
                LOG_INFO(log, "Closed connections.");

            bool joined_refresh_tasks = global_context->getRefreshSet().joinBackgroundTasks(shutdown_deadline);
            bool joined_background_queries = global_context->getBackgroundQueryPool().waitUntil(shutdown_deadline);

            if (!joined_background_queries)
                LOG_WARNING(log, "{} background queries remain.", global_context->getBackgroundQueryPool().active());

            dns_cache_updater.reset();

            /// killAllQueries() and the waits above can have driven a cancelled backup to a final
            /// status, and then the normal teardown is able to complete. Sound because this is false
            /// only where new operations are already refused, so the unfinished set cannot grow.
            if (!backups_finished)
                backups_finished = !global_context->hasUnfinishedBackupsAndRestores();

            if (current_connections || !joined_refresh_tasks || !joined_background_queries || !backups_finished)
            {
                /// There is no better way to force connections to close in Poco.
                /// Otherwise connection handlers will continue to live
                /// (they are effectively dangling objects, but they use global thread pool
                ///  and global thread pool destructor will wait for threads, preventing server shutdown).

                /// Dump coverage here, because std::atexit callback would not be called.
                dumpCoverageReportIfPossible();
                LOG_WARNING(log, "Will shutdown forcefully.");
                /// No leak check: remaining connection handlers, refresh tasks or background queries still own memory
                /// it would report as leaked. Reported, because here stderr is the server log.
                safeExit(0, LeakCheck::SkipAndReport);
            }

            /// Before the AsyncLoader stops, because a restore can create a database, and after the forced exit, because stopping
            /// the DDL worker waits for the task it runs, which the kill cancels unless it is such a part.
            if (wait_backups_and_restores)
            {
                global_context->getProcessList().killAllQueries(/* except_ddl_backups_and_restores= */ true);
                global_context->stopDDLWorkerAndWaitAllBackupsAndRestores();
            }
        });

        {
            std::lock_guard lock(servers_lock);

            /// Restore the startup log level overrides before accepting connections,
            /// so that no requests are served with an elevated (startup) log level.
            /// This must happen before server.start() because the config reload callback
            /// (ConfigReloader) reads from config() which includes the writable layer
            /// where startup level overrides are stored.
            if (should_restore_default_logger_level)
            {
                config().setString("logger.level", default_logger_level_config);
                Loggers::updateLevels(config(), logger());
                LOG_INFO(log, "Restored default logger level to {}", default_logger_level_config);
            }

            if (should_restore_console_log_level)
            {
                /// If the level was unset just remove the override so the default can be set via
                /// Loggers::updateLevels again; otherwise restore the configured value.
                if (console_log_level_was_set)
                {
                    config().setString("logger.console_log_level", original_console_log_level_config);
                    Loggers::updateLevels(config(), logger());
                    LOG_INFO(log, "Restored console logger level to {}", original_console_log_level_config);
                }
                else
                {
                    config().remove("logger.console_log_level");
                    Loggers::updateLevels(config(), logger());
                    LOG_INFO(log, "Restored console logger level to logger.level");
                }
            }

            startServers(servers, listen_try, log);
            for (const auto & server : servers)
            {
                if (server.bindsOnStart())
                    global_context->registerServerPort(server.getPortName(), static_cast<UInt16>(config().getInt(server.getPortName())));
            }
            if (servers.empty())
                throw Exception(ErrorCodes::NO_ELEMENTS_IN_CONFIG,
                    "No servers started (add valid listen_host and 'tcp_port' or 'http_port' to configuration file.)");

            global_context->setServerCompletelyStarted();
            LOG_INFO(log, "Ready for connections.");
        }

        startup_watch.stop();
        ProfileEvents::increment(ProfileEvents::ServerStartupMilliseconds, startup_watch.elapsedMilliseconds());

        CannotAllocateThreadFaultInjector::setFaultProbability(server_settings[ServerSetting::cannot_allocate_thread_fault_injection_probability]);

        try
        {
            global_context->startClusterDiscovery();
        }
        catch (...)
        {
            tryLogCurrentException(log, "Caught exception while starting cluster discovery");
        }

#if defined(OS_LINUX)
        /// Tell the service manager that service startup is finished.
        /// NOTE: the parent clickhouse-watchdog process must do systemdNotify("MAINPID={}\n", child_pid); before
        /// the child process notifies 'READY=1'.
        systemdNotify("READY=1\n");
#endif

        std::vector<std::unique_ptr<MetricsTransmitter>> metrics_transmitters;
        for (const auto & graphite_key : DB::getMultipleKeysFromConfig(config(), "", "graphite"))
        {
            metrics_transmitters.emplace_back(std::make_unique<MetricsTransmitter>(
                global_context->getConfigRef(), graphite_key, *async_metrics));
        }

        waitForTerminationRequest();
    }

    return Application::EXIT_OK;
}
catch (...)
{
    /// Poco does not provide stacktrace.
    tryLogCurrentException("Application");
    auto code = getCurrentExceptionCode();
    return static_cast<UInt8>(code) ? code : -1;
}

namespace
{

/// Maps a `<protocols>` stack onto the ServerType of its innermost layer, so that a composable
/// protocol is reported under the same metrics as the equivalent built-in port.
ServerType::Type protocolStackType(const std::string & leaf_type, bool is_secure, bool has_proxy)
{
    if (leaf_type == "tcp")
    {
        if (is_secure)
            return ServerType::Type::TCP_SECURE;
        return has_proxy ? ServerType::Type::TCP_WITH_PROXY : ServerType::Type::TCP;
    }
    if (leaf_type == "http")
        return is_secure ? ServerType::Type::HTTPS : ServerType::Type::HTTP;
    if (leaf_type == "interserver")
        return is_secure ? ServerType::Type::INTERSERVER_HTTPS : ServerType::Type::INTERSERVER_HTTP;
    if (leaf_type == "mysql")
        return ServerType::Type::MYSQL;
    if (leaf_type == "postgres")
        return ServerType::Type::POSTGRESQL;
    if (leaf_type == "prometheus")
        return ServerType::Type::PROMETHEUS;

    return ServerType::Type::END;
}

/// Walk a composable protocol's `impl` chain and return the effective `default_session_user`:
/// the value closest to the endpoint wins. Used both when the protocol stack is built and when it is
/// decided whether a configuration reload has to restart the endpoint.
/// Returns an empty optional if no module in the chain sets it (the handlers then fall back to the
/// `default_session_user` server setting).
std::optional<String> getEffectiveDefaultSessionUser(const Poco::Util::AbstractConfiguration & config, const std::string & protocol)
{
    std::string conf_name = protocol;
    std::string prefix = protocol + ".";
    std::unordered_set<std::string> pset {conf_name};
    while (true)
    {
        if (config.has(prefix + "default_session_user"))
            return config.getString(prefix + "default_session_user");

        if (!config.has(prefix + "impl"))
            return {};

        conf_name = "protocols." + config.getString(prefix + "impl");
        prefix = conf_name + ".";

        /// A malformed loop is rejected when the stack is built; here we just stop to avoid spinning.
        if (!pset.insert(conf_name).second)
            return {};
    }
}

/// Resolve the `http_handlers`-style section a composable `http` endpoint serves: the
/// `handlers` key of the module carrying `type = http`, found by walking the `impl` chain
/// (mirrors `buildProtocolStackFromConfig`). Returns an empty optional when the chain does
/// not reach an `http` module in this configuration (e.g. the endpoint did not exist or had
/// a different type before a reload).
std::optional<String> resolveHTTPHandlersKey(const Poco::Util::AbstractConfiguration & config, const std::string & protocol)
{
    std::string conf_name = protocol;
    std::string prefix = protocol + ".";
    std::unordered_set<std::string> pset {conf_name};
    while (true)
    {
        if (config.getString(prefix + "type", "") == "http")
            return config.getString(conf_name + ".handlers", "http_handlers");

        if (!config.has(prefix + "impl"))
            return {};

        conf_name = "protocols." + config.getString(prefix + "impl");
        prefix = conf_name + ".";

        /// A malformed loop is rejected when the stack is built; here we just stop to avoid spinning.
        if (!pset.insert(conf_name).second)
            return {};
    }
}

/// The `<http_handlers>`-style sections the HTTP listeners of a configuration serve: the default
/// `http_handlers` for `http_port` and `https_port`, and the section each composable `http` endpoint
/// references. A section no listener serves is left out, so that validating a configuration accepts
/// exactly what starting the server with it would.
Strings servedHTTPHandlersKeys(const Poco::Util::AbstractConfiguration & config)
{
    std::unordered_set<String> keys;

    if (config.has("http_port") || config.has("https_port"))
        keys.insert("http_handlers");

    if (config.has("protocols"))
    {
        Poco::Util::AbstractConfiguration::Keys protocols;
        config.keys("protocols", protocols);
        for (const auto & protocol : protocols)
        {
            if (auto handlers_key = resolveHTTPHandlersKey(config, "protocols." + protocol))
                keys.insert(*handlers_key);
        }
    }

    return {keys.begin(), keys.end()};
}

/// Whether a listener of this configuration serves the `prometheus` section on a port of its own: the
/// standalone `prometheus.port` listener, or a composable `type = prometheus` endpoint (whose type is
/// found by walking the `impl` chain, as in `buildProtocolStackFromConfig`).
bool hasPrometheusListener(const Poco::Util::AbstractConfiguration & config)
{
    if (config.getInt("prometheus.port", 0))
        return true;

    if (!config.has("protocols"))
        return false;

    Poco::Util::AbstractConfiguration::Keys protocols;
    config.keys("protocols", protocols);

    for (const auto & protocol : protocols)
    {
        std::string conf_name = "protocols." + protocol;
        std::string prefix = conf_name + ".";
        std::unordered_set<std::string> pset {conf_name};
        while (true)
        {
            if (config.getString(prefix + "type", "") == "prometheus")
                return true;

            if (!config.has(prefix + "impl"))
                break;

            conf_name = "protocols." + config.getString(prefix + "impl");
            prefix = conf_name + ".";

            /// A malformed loop is rejected when the stack is built; here we just stop to avoid spinning.
            if (!pset.insert(conf_name).second)
                break;
        }
    }

    return false;
}

/// Whether a non-keeper `prometheus` endpoint (serving the global `prometheus` section)
/// consults the default session user. Only the time-series handlers without a fixed `user`
/// authenticate; the plain metrics exposition served without a `handlers` section does not.
bool prometheusHandlersConsumeDefaultSessionUser(const Poco::Util::AbstractConfiguration & config)
{
    if (!config.has("prometheus.handlers"))
        return false;

    Poco::Util::AbstractConfiguration::Keys keys;
    config.keys("prometheus.handlers", keys);
    for (const auto & key : keys)
    {
        const std::string handler_prefix = "prometheus.handlers." + key + ".handler";
        const std::string type = config.getString(handler_prefix + ".type", "");
        /// The metrics exposition aliases accepted by `parseHandlerType`.
        if (type == "prometheus" || type == "metrics" || type == "expose_metrics"
            || type == "prometheus_metrics" || type == "prometheus_expose_metrics")
            continue;
        if (!config.has(handler_prefix + ".user"))
            return true;
    }
    return false;
}

}

std::unique_ptr<TCPProtocolStackFactory> Server::buildProtocolStackFromConfig(
    const Poco::Util::AbstractConfiguration & config,
    const ServerSettings & server_settings,
    const std::string & protocol,
    Poco::Net::HTTPServerParams::Ptr http_params,
    AsynchronousMetrics & async_metrics,
    bool & is_secure,
    ServerType::Type & protocol_type)
{
    /// The default session user for the endpoint: the `default_session_user` key looked up
    /// from the endpoint's protocol module towards the referenced (`impl`) modules; the value
    /// closest to the endpoint wins. If not set anywhere in the chain, the handlers fall back
    /// to the `default_session_user` server setting.
    /// It is resolved for the whole `impl` chain up front, because a module can be both typed
    /// (so its handler factory is created while walking the chain) and inherit the setting from
    /// a module it references, which the walk reaches only later.
    const std::optional<String> default_session_user = getEffectiveDefaultSessionUser(config, "protocols." + protocol);

    auto create_factory = [&](const std::string & type, const std::string & conf_name) -> TCPServerConnectionFactory::Ptr
    {
        if (type == "tcp")
            return TCPServerConnectionFactory::Ptr(new TCPHandlerFactory(*this, false, false, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes, default_session_user));

        if (type == "tls")
#if USE_SSL
            return TCPServerConnectionFactory::Ptr(new TLSHandlerFactory(*this, conf_name));
#else
            throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support.");
#endif

        if (type == "proxy1")
            return TCPServerConnectionFactory::Ptr(new ProxyV1HandlerFactory(*this, conf_name));
        if (type == "mysql")
            return TCPServerConnectionFactory::Ptr(new MySQLHandlerFactory(*this, server_settings[ServerSetting::mysql_require_secure_transport], ProfileEvents::InterfaceMySQLReceiveBytes, ProfileEvents::InterfaceMySQLSendBytes, default_session_user));
        if (type == "postgres")
#if USE_SSL
            return TCPServerConnectionFactory::Ptr(new PostgreSQLHandlerFactory(*this, server_settings[ServerSetting::postgresql_require_secure_transport], conf_name + ".", ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes, default_session_user));
#else
            return TCPServerConnectionFactory::Ptr(new PostgreSQLHandlerFactory(*this, server_settings[ServerSetting::postgresql_require_secure_transport], ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes, default_session_user));
#endif
        if (type == "http")
        {
            /// Allow custom http_handlers configuration for this protocol endpoint.
            /// E.g. <protocols><my_http><type>http</type><handlers>http_handlers_alt</handlers></my_http></protocols>
            String handlers_config_key;
            if (config.has(conf_name + ".handlers"))
                handlers_config_key = config.getString(conf_name + ".handlers");
            return TCPServerConnectionFactory::Ptr(
                new HTTPServerConnectionFactory(httpContext(), http_params, createHandlerFactory(*this, config, async_metrics, "HTTPHandler-factory", handlers_config_key, protocol, default_session_user), ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes)
            );
        }
        if (type == "prometheus")
        {
            const std::string handler_name = server_settings[ServerSetting::prometheus_keeper_metrics_only] ? "KeeperPrometheusHandler-factory" : "PrometheusHandler-factory";
            return TCPServerConnectionFactory::Ptr(
                new HTTPServerConnectionFactory(httpContext(), http_params, createHandlerFactory(*this, config, async_metrics, handler_name, /* http_handlers_key= */ {}, protocol, default_session_user), ProfileEvents::InterfacePrometheusReceiveBytes, ProfileEvents::InterfacePrometheusSendBytes)
            );
        }
        if (type == "interserver")
            return TCPServerConnectionFactory::Ptr(
                new HTTPServerConnectionFactory(httpContext(), http_params, createHandlerFactory(*this, config, async_metrics, "InterserverIOHTTPHandler-factory"), ProfileEvents::InterfaceInterserverReceiveBytes, ProfileEvents::InterfaceInterserverSendBytes)
            );

        throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol configuration error, unknown protocol name '{}'", type);
    };

    std::string conf_name = "protocols." + protocol;
    std::string prefix = conf_name + ".";
    std::unordered_set<std::string> pset {conf_name};

    const bool is_introspection = isIntrospectionProtocol(config, protocol);
    std::string innermost_type;

    auto stack = std::make_unique<TCPProtocolStackFactory>(*this, conf_name, is_introspection);

    bool has_interserver = false;

    /// `tls` and `proxy1` only wrap another protocol; the innermost layer of the chain is
    /// what the server actually speaks, and it determines which metrics are reported for it.
    std::string leaf_type;
    bool has_proxy = false;

    while (true)
    {
        // if there is no "type" - it's a reference to another protocol and this is just an endpoint
        if (config.has(prefix + "type"))
        {
            std::string type = config.getString(prefix + "type");
            if (type == "tls")
            {
                if (is_secure)
                    throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' contains more than one TLS layer", protocol);
                is_secure = true;
            }
            else if (type == "proxy1")
                has_proxy = true;
            else
                leaf_type = type;

            if (type == "interserver")
                has_interserver = true;

            if ((type == "tls" || type == "postgres") && !Poco::trim(config.getString(prefix + "cipherSuites", "")).empty())
            {
                const auto private_key_file = config.getString(prefix + "privateKeyFile", "");
                if (private_key_file.empty() || config.getString(prefix + "certificateFile", private_key_file).empty())
                {
                    ///  builds a context of its own for an ACME certificate too, but it reads the
                    /// layer-local TLS options only together with a key pair, so the value would be dropped.
                    if (type == "tls" && config.has("acme"))
                        throw Exception(
                            ErrorCodes::INVALID_CONFIG_PARAMETER,
                            "Protocol '{}': 'cipherSuites' in '{}' is not applied to a layer served with an ACME certificate; "
                            "set it in the 'openSSL.server' section or give the layer its own 'privateKeyFile'", protocol, conf_name);
                    throw Exception(
                        ErrorCodes::INVALID_CONFIG_PARAMETER,
                        "Protocol '{}': 'cipherSuites' in '{}' requires a 'privateKeyFile' (and 'certificateFile', if separate) "
                        "in the same section, without them it cannot be applied", protocol, conf_name);
                }
            }

            if (is_introspection && type != "tcp" && type != "tls" && type != "proxy1")
                throw Exception(
                    ErrorCodes::INVALID_CONFIG_PARAMETER,
                    "Introspection protocol '{}' contains a '{}' layer, but only 'tcp' optionally wrapped "
                    "into 'tls' and 'proxy1' layers is supported", protocol, type);

            innermost_type = type;
            stack->append(create_factory(type, conf_name));
        }

        if (!config.has(prefix + "impl"))
            break;

        conf_name = "protocols." + config.getString(prefix + "impl");
        prefix = conf_name + ".";

        if (!pset.insert(conf_name).second)
            throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' configuration contains a loop on '{}'", protocol, conf_name);
    }

    /// Interserver connections are authenticated by the cluster secret and the initial user
    /// and never use the default session user, so such a configuration is an error.
    if (has_interserver && default_session_user)
        throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER,
            "Protocol '{}': 'default_session_user' cannot be used with the 'interserver' protocol", protocol);

    if (is_introspection && innermost_type != "tcp")
        throw Exception(
            ErrorCodes::INVALID_CONFIG_PARAMETER,
            "Introspection protocol '{}' must end in a 'tcp' layer", protocol);

    protocol_type = protocolStackType(leaf_type, is_secure, has_proxy);

    return stack;
}

HTTPContextPtr Server::httpContext() const
{
    return std::make_shared<HTTPContext>(context());
}

void Server::createServers(
    Poco::Util::AbstractConfiguration & config,
    const ServerSettings & server_settings,
    const Strings & listen_hosts,
    bool listen_try,
    Poco::ThreadPool & server_pool,
    AsynchronousMetrics & async_metrics,
    std::vector<ProtocolServerAdapter> & servers,
    bool start_servers,
    const ServerType & server_type,
    bool only_introspection_protocols)
{
    const Settings & settings = global_context->getSettingsRef();
#if USE_GRPC || USE_ARROWFLIGHT
    const auto grpc_listen_hosts = getGRPCListenHosts(listen_hosts);
#endif

    Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams;
    http_params->setTimeout(settings[Setting::http_receive_timeout]);
    http_params->setKeepAliveTimeout(global_context->getServerSettings()[ServerSetting::keep_alive_timeout]);
    http_params->setMaxKeepAliveRequests(static_cast<int>(global_context->getServerSettings()[ServerSetting::max_keep_alive_requests]));
    http_params->setMaxQueued(server_settings[ServerSetting::listen_backlog]);

    Poco::Util::AbstractConfiguration::Keys protocols;
    config.keys("protocols", protocols);

    const TCPServerConnectionFilter::Ptr & connection_filter = new TCPServerConnectionFilter{[&]()
    {
        return !ProfileEvents::checkCPUOverload(global_context->getServerSettings()[ServerSetting::os_cpu_busy_time_threshold],
                global_context->getMinOSCPUWaitTimeRatioToDropConnection(),
                global_context->getMaxOSCPUWaitTimeRatioToDropConnection(),
                /*should_throw*/ false);
    }};

    for (const auto & protocol : protocols)
    {
        const bool is_introspection = isIntrospectionProtocol(config, protocol);
        if (is_introspection != only_introspection_protocols)
            continue;

        if (!server_type.shouldStart(ServerType::Type::CUSTOM, protocol))
            continue;

        std::string prefix = "protocols." + protocol + ".";
        std::string port_name = prefix + "port";
        std::string description {"<undefined> protocol"};
        if (config.has(prefix + "description"))
            description = config.getString(prefix + "description");

        if (!config.has(prefix + "port"))
            continue;

        std::vector<std::string> hosts;
        if (config.has(prefix + "host"))
            hosts.push_back(config.getString(prefix + "host"));
        else
            hosts = listen_hosts;

        for (const auto & host : hosts)
        {
            bool is_secure = false;
            auto protocol_type = ServerType::Type::END;
            auto stack = buildProtocolStackFromConfig(
                config, server_settings, protocol, http_params, async_metrics, is_secure, protocol_type);

            if (stack->empty())
                throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' stack empty", protocol);

            createServer(config, host, port_name.c_str(), listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                Poco::Net::ServerSocket socket;
                auto address = socketBindListen(server_settings, socket, host, port, is_secure);
                socket.setReceiveTimeout(settings[Setting::receive_timeout]);
                socket.setSendTimeout(settings[Setting::send_timeout]);

                return ProtocolServerAdapter(
                    host,
                    port_name.c_str(),
                    protocol_type,
                    description + ": " + address.toString(),
                    std::make_unique<TCPServer>(
                        stack.release(),
                        server_pool,
                        socket,
                        makeServerParams(server_settings),
                        is_introspection ? TCPServerConnectionFilter::Ptr() : connection_filter));
            });
        }
    }

    /// Introspection ports can only be configured in the protocols config section.
    if (only_introspection_protocols)
        return;

    for (const auto & listen_host : listen_hosts)
    {
        const char * port_name = nullptr;

        if (server_type.shouldStart(ServerType::Type::HTTP))
        {
            /// HTTP
            port_name = "http_port";
            /// Handler configuration is parsed outside the callback: anything the callback throws is
            /// reported as a listener bind failure, and only logged when `listen_try` is set. The port
            /// check matches the early return in `createServer`.
            if (!config.getString(port_name, "").empty())
            {
                auto handler_factory = createHandlerFactory(*this, config, async_metrics, "HTTPHandler-factory");
                createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
                {
                    Poco::Net::ServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port);
                    socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                    socket.setSendTimeout(settings[Setting::http_send_timeout]);

                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::HTTP,
                        "http://" + address.toString(),
                        std::make_unique<HTTPServer>(
                            httpContext(), handler_factory, server_pool, socket, http_params, connection_filter, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes));
                });
            }
        }

        if (server_type.shouldStart(ServerType::Type::HTTPS))
        {
            /// HTTPS
            port_name = "https_port";
            if (!config.getString(port_name, "").empty())
            {
#if USE_SSL
                auto handler_factory = createHandlerFactory(*this, config, async_metrics, "HTTPSHandler-factory");
#endif
                createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
                {
#if USE_SSL
                    Poco::Net::SecureServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                    socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                    socket.setSendTimeout(settings[Setting::http_send_timeout]);
                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::HTTPS,
                        "https://" + address.toString(),
                        std::make_unique<HTTPServer>(
                            httpContext(), handler_factory, server_pool, socket, http_params, connection_filter, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes));
#else
                    UNUSED(port);
                    throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "HTTPS protocol is disabled because Poco library was built without NetSSL support.");
#endif
                });
            }
        }

        if (server_type.shouldStart(ServerType::Type::TCP))
        {
            /// TCP
            port_name = "tcp_port";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                Poco::Net::ServerSocket socket;
                auto address = socketBindListen(server_settings, socket, listen_host, port);
                socket.setReceiveTimeout(settings[Setting::receive_timeout]);
                socket.setSendTimeout(settings[Setting::send_timeout]);
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::TCP,
                    "native protocol (tcp): " + address.toString(),
                    std::make_unique<TCPServer>(
                        new TCPHandlerFactory(*this, /* secure */ false, /* proxy protocol */ false, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes),
                        server_pool,
                        socket,
                        makeServerParams(server_settings),
                        connection_filter));
            });
        }

        if (server_type.shouldStart(ServerType::Type::TCP_WITH_PROXY))
        {
            /// TCP with PROXY protocol, see https://github.com/wolfeidau/proxyv2/blob/master/docs/proxy-protocol.txt
            port_name = "tcp_with_proxy_port";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                Poco::Net::ServerSocket socket;
                auto address = socketBindListen(server_settings, socket, listen_host, port);
                socket.setReceiveTimeout(settings[Setting::receive_timeout]);
                socket.setSendTimeout(settings[Setting::send_timeout]);
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::TCP_WITH_PROXY,
                    "native protocol (tcp) with PROXY: " + address.toString(),
                    std::make_unique<TCPServer>(
                        new TCPHandlerFactory(*this, /* secure */ false, /* proxy protocol */ true, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes),
                        server_pool,
                        socket,
                        makeServerParams(server_settings),
                        connection_filter));
            });
        }

#if USE_ARROWFLIGHT
        if (server_type.shouldStart(ServerType::Type::ARROW_FLIGHT)
            && std::ranges::find(grpc_listen_hosts, listen_host) != grpc_listen_hosts.end())
        {
            port_name = "arrowflight_port";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                /// Do not bind a Poco socket here: gRPC owns the listening socket of an Arrow Flight
                /// server and binds it in `start`. A pre-bind would both bind the address twice and
                /// defeat the family-agnostic handling of a wildcard address by gRPC, which falls
                /// back to `0.0.0.0` for `::` on a host without IPv6.
                auto address = makeSocketAddress(listen_host, port, &logger());
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::ARROW_FLIGHT,
                    "Arrow Flight compatibility protocol: " + address.toString(),
                    std::unique_ptr<IGRPCServer>(new ArrowFlightServer(*this, address)),
                    true);
            });
        }
#endif

        if (server_type.shouldStart(ServerType::Type::TCP_SECURE))
        {
            /// TCP with SSL
            port_name = "tcp_port_secure";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
    #if USE_SSL
                Poco::Net::SecureServerSocket socket;
                auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                socket.setReceiveTimeout(settings[Setting::receive_timeout]);
                socket.setSendTimeout(settings[Setting::send_timeout]);
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::TCP_SECURE,
                    "secure native protocol (tcp_secure): " + address.toString(),
                    std::make_unique<TCPServer>(
                        new TCPHandlerFactory(*this, /* secure */ true, /* proxy protocol */ false, ProfileEvents::InterfaceNativeReceiveBytes, ProfileEvents::InterfaceNativeSendBytes),
                        server_pool,
                        socket,
                        makeServerParams(server_settings),
                        connection_filter));
    #else
                UNUSED(port);
                throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support.");
    #endif
            });
        }

        if (server_type.shouldStart(ServerType::Type::TCP_SSH))
        {
            port_name = "tcp_ssh_port";
            createServer(
                config,
                listen_host,
                port_name,
                listen_try,
                start_servers,
                servers,
                [&](UInt16 port) -> ProtocolServerAdapter
                {
#if USE_SSH && defined(OS_LINUX)
                    Poco::Net::ServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ false);
                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::TCP_SSH,
                        "SSH PTY: " + address.toString(),
                        std::make_unique<TCPServer>(
                            new SSHPtyHandlerFactory(*this, config),
                            server_pool,
                            socket,
                            makeServerParams(server_settings),
                            connection_filter));
#else
                UNUSED(port);
                throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSH protocol is disabled because ClickHouse has been built without libssh or is running on a non-Linux platform");
#endif
                });
        }

        if (server_type.shouldStart(ServerType::Type::MYSQL))
        {
            port_name = "mysql_port";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                Poco::Net::ServerSocket socket;
                auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                socket.setReceiveTimeout(Poco::Timespan());
                socket.setSendTimeout(settings[Setting::send_timeout]);
                bool secure_required = server_settings[ServerSetting::mysql_require_secure_transport];
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::MYSQL,
                    "MySQL compatibility protocol: " + address.toString(),
                    std::make_unique<TCPServer>(
                         new MySQLHandlerFactory(*this, secure_required, ProfileEvents::InterfaceMySQLReceiveBytes, ProfileEvents::InterfaceMySQLSendBytes),
                         server_pool,
                         socket,
                         makeServerParams(server_settings),
                         connection_filter));
            });
        }

        if (server_type.shouldStart(ServerType::Type::POSTGRESQL))
        {
            port_name = "postgresql_port";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                Poco::Net::ServerSocket socket;
                auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                socket.setReceiveTimeout(Poco::Timespan());
                socket.setSendTimeout(settings[Setting::send_timeout]);
                bool secure_required = server_settings[ServerSetting::postgresql_require_secure_transport];
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::POSTGRESQL,
                    "PostgreSQL compatibility protocol: " + address.toString(),
#if USE_SSL
                    std::make_unique<TCPServer>(
                         new PostgreSQLHandlerFactory(*this, secure_required, Poco::Net::SSLManager::CFG_SERVER_PREFIX, ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes),
                         server_pool,
                         socket,
                         makeServerParams(server_settings),
                         connection_filter));
#else
                    std::make_unique<TCPServer>(
                         new PostgreSQLHandlerFactory(*this, secure_required, ProfileEvents::InterfacePostgreSQLReceiveBytes, ProfileEvents::InterfacePostgreSQLSendBytes),
                         server_pool,
                         socket,
                         makeServerParams(server_settings),
                         connection_filter));
#endif
            });
        }

#if USE_GRPC
        if (server_type.shouldStart(ServerType::Type::GRPC)
            && std::ranges::find(grpc_listen_hosts, listen_host) != grpc_listen_hosts.end())
        {
            port_name = "grpc_port";
            createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                auto server_address = makeSocketAddress(listen_host, port, &logger());
                return ProtocolServerAdapter(
                    listen_host,
                    port_name,
                    ServerType::Type::GRPC,
                    "gRPC protocol: " + server_address.toString(),
                    std::make_unique<GRPCServer>(*this, server_address));
            });
        }
#endif
        if (server_type.shouldStart(ServerType::Type::PROMETHEUS))
        {
            /// Prometheus (if defined and not setup yet with http_port)
            port_name = "prometheus.port";

            const char * handler_name = server_settings[ServerSetting::prometheus_keeper_metrics_only] ? "KeeperPrometheusHandler-factory" : "PrometheusHandler-factory";

            if (!config.getString(port_name, "").empty())
            {
                auto handler_factory = createHandlerFactory(*this, config, async_metrics, handler_name);
                createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
                {
                    Poco::Net::ServerSocket socket;
                    auto address = socketBindListen(server_settings, socket, listen_host, port);
                    socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                    socket.setSendTimeout(settings[Setting::http_send_timeout]);
                    return ProtocolServerAdapter(
                        listen_host,
                        port_name,
                        ServerType::Type::PROMETHEUS,
                        "Prometheus: http://" + address.toString(),
                        std::make_unique<HTTPServer>(
                            httpContext(), handler_factory, server_pool, socket, http_params, nullptr, ProfileEvents::InterfacePrometheusReceiveBytes, ProfileEvents::InterfacePrometheusSendBytes));
                });
            }
        }

        bool has_catalog_port = !config.getString("iceberg_rest_catalog.port", "").empty();
        bool has_catalog_secure_port = !config.getString("iceberg_rest_catalog.secure_port", "").empty();
        if (server_type.shouldStart(ServerType::Type::ICEBERG_REST_CATALOG)
            && (has_catalog_port || has_catalog_secure_port))
        {
            auto is_listening = [&](const char * name)
            {
                return std::any_of(servers.begin(), servers.end(), [&](const auto & server)
                {
                    return !server.isStopping() && server.getListenHost() == listen_host && server.getPortName() == name;
                });
            };

            /// Both listeners share one handler factory. Create it only if some configured port is not listening yet.
            const bool already_listening = (!has_catalog_port || is_listening("iceberg_rest_catalog.port"))
                && (!has_catalog_secure_port || is_listening("iceberg_rest_catalog.secure_port"));

            HTTPRequestHandlerFactoryPtr handler_factory;
            if (!already_listening)
            {
                try
                {
                    handler_factory = createIcebergRESTCatalogHandlerFactory(*this, config);
                }
                catch (...)
                {
                    LOG_ERROR(&logger(), "Not starting the Iceberg REST catalog server: {}", getCurrentExceptionMessage(/*with_stacktrace*/ false));
                }
            }

            if (handler_factory)
            {
                if (has_catalog_port)
                {
                    port_name = "iceberg_rest_catalog.port";
                    createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
                    {
                        Poco::Net::ServerSocket socket;
                        auto address = socketBindListen(server_settings, socket, listen_host, port);
                        socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                        socket.setSendTimeout(settings[Setting::http_send_timeout]);
                        return ProtocolServerAdapter(
                            listen_host,
                            port_name,
                            ServerType::Type::ICEBERG_REST_CATALOG,
                            "Iceberg REST catalog: http://" + address.toString(),
                            std::make_unique<HTTPServer>(
                                httpContext(), handler_factory, server_pool, socket, http_params, nullptr, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes));
                    });
                }

                /// The TLS listener uses the certificate from the <openSSL><server> section.
                if (has_catalog_secure_port)
                {
                    port_name = "iceberg_rest_catalog.secure_port";
                    createServer(config, listen_host, port_name, listen_try, start_servers, servers, [&]([[maybe_unused]] UInt16 port) -> ProtocolServerAdapter
                    {
#if USE_SSL
                        Poco::Net::SecureServerSocket socket;
                        auto address = socketBindListen(server_settings, socket, listen_host, port, /* secure = */ true);
                        socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                        socket.setSendTimeout(settings[Setting::http_send_timeout]);
                        return ProtocolServerAdapter(
                            listen_host,
                            port_name,
                            ServerType::Type::ICEBERG_REST_CATALOG,
                            "Iceberg REST catalog: https://" + address.toString(),
                            std::make_unique<HTTPServer>(
                                httpContext(), handler_factory, server_pool, socket, http_params, nullptr, ProfileEvents::InterfaceHTTPReceiveBytes, ProfileEvents::InterfaceHTTPSendBytes));
#else
                        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "HTTPS for the Iceberg REST catalog is disabled because Poco library was built without NetSSL support.");
#endif
                    });
                }
            }
        }
    }
}

void Server::createInterserverServers(
    Poco::Util::AbstractConfiguration & config,
    const ServerSettings & server_settings,
    const Strings & interserver_listen_hosts,
    bool listen_try,
    Poco::ThreadPool & server_pool,
    AsynchronousMetrics & async_metrics,
    std::vector<ProtocolServerAdapter> & servers,
    bool start_servers,
    const ServerType & server_type)
{
    const Settings & settings = global_context->getSettingsRef();

    Poco::Net::HTTPServerParams::Ptr http_params = new Poco::Net::HTTPServerParams;
    http_params->setTimeout(settings[Setting::http_receive_timeout]);
    http_params->setKeepAliveTimeout(global_context->getServerSettings()[ServerSetting::keep_alive_timeout]);
    http_params->setMaxQueued(server_settings[ServerSetting::listen_backlog]);

    /// Now iterate over interserver_listen_hosts
    for (const auto & interserver_listen_host : interserver_listen_hosts)
    {
        const char * port_name = nullptr;

        if (server_type.shouldStart(ServerType::Type::INTERSERVER_HTTP))
        {
            /// Interserver IO HTTP
            port_name = "interserver_http_port";
            createServer(config, interserver_listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
                Poco::Net::ServerSocket socket;
                auto address = socketBindListen(server_settings, socket, interserver_listen_host, port);
                socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                socket.setSendTimeout(settings[Setting::http_send_timeout]);
                return ProtocolServerAdapter(
                    interserver_listen_host,
                    port_name,
                    ServerType::Type::INTERSERVER_HTTP,
                    "replica communication (interserver): http://" + address.toString(),
                    std::make_unique<HTTPServer>(
                        httpContext(),
                        createHandlerFactory(*this, config, async_metrics, "InterserverIOHTTPHandler-factory"),
                        server_pool,
                        socket,
                        http_params,
                        nullptr,
                        ProfileEvents::InterfaceInterserverReceiveBytes,
                        ProfileEvents::InterfaceInterserverSendBytes));
            });
        }

        if (server_type.shouldStart(ServerType::Type::INTERSERVER_HTTPS))
        {
            port_name = "interserver_https_port";
            createServer(config, interserver_listen_host, port_name, listen_try, start_servers, servers, [&](UInt16 port) -> ProtocolServerAdapter
            {
#if USE_SSL
                Poco::Net::SecureServerSocket socket;
                auto address = socketBindListen(server_settings, socket, interserver_listen_host, port, /* secure = */ true);
                socket.setReceiveTimeout(settings[Setting::http_receive_timeout]);
                socket.setSendTimeout(settings[Setting::http_send_timeout]);
                return ProtocolServerAdapter(
                    interserver_listen_host,
                    port_name,
                    ServerType::Type::INTERSERVER_HTTPS,
                    "secure replica communication (interserver): https://" + address.toString(),
                    std::make_unique<HTTPServer>(
                        httpContext(),
                        createHandlerFactory(*this, config, async_metrics, "InterserverIOHTTPSHandler-factory"),
                        server_pool,
                        socket,
                        http_params,
                        nullptr,
                        ProfileEvents::InterfaceInterserverReceiveBytes,
                        ProfileEvents::InterfaceInterserverSendBytes));
#else
                UNUSED(port);
                throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "SSL support for TCP protocol is disabled because Poco library was built without NetSSL support.");
#endif
            });
        }
    }
}

void Server::stopServers(
    std::vector<ProtocolServerAdapter> & servers,
    const ServerType & server_type) const
{
    DB::stopServers(servers, server_type, &logger());
}

void Server::updateServers(
    Poco::Util::AbstractConfiguration & config,
    const ServerSettings & server_settings,
    Poco::ThreadPool & server_pool,
    AsynchronousMetrics & async_metrics,
    std::vector<ProtocolServerAdapter> & servers,
    std::vector<ProtocolServerAdapter> & servers_to_start_before_tables)
{
    LoggerRawPtr log = &logger();

    const auto listen_hosts = getListenHosts(config);
    const auto interserver_listen_hosts = getInterserverListenHosts(config);
    const auto listen_try = getListenTry(config, server_settings);
#if USE_GRPC || USE_ARROWFLIGHT
    const auto grpc_listen_hosts = getGRPCListenHosts(listen_hosts);
#endif

    /// Remove servers once all their connections are closed
    auto check_server = [&log](const char prefix[], auto & server)
    {
        if (!server.isStopping())
            return false;
        size_t current_connections = server.currentConnections();
        LOG_DEBUG(log, "Server {}{}: {} ({} connections)",
            server.getDescription(),
            prefix,
            !current_connections ? "finished" : "waiting",
            current_connections);
        return !current_connections;
    };

    std::erase_if(servers, std::bind_front(check_server, " (from one of previous reload)"));

    Poco::Util::AbstractConfiguration & previous_config = latest_config ? *latest_config : config;

    std::vector<ProtocolServerAdapter *> all_servers;
    all_servers.reserve(servers.size() + servers_to_start_before_tables.size());
    for (auto & server : servers)
        all_servers.push_back(&server);

    for (auto & server : servers_to_start_before_tables)
        all_servers.push_back(&server);

    for (auto * server : all_servers)
    {
        if (server->supportsRuntimeReconfiguration() && !server->isStopping())
        {
            std::string port_name = server->getPortName();
            bool has_host = false;
            bool is_http = false;
            bool is_non_keeper_prometheus = false;
            bool is_prometheus = false;
            bool default_session_user_changed = false;
            String handlers_key = "http_handlers";
            String previous_handlers_key = handlers_key;
            if (port_name.starts_with("protocols."))
            {
                std::string protocol = port_name.substr(0, port_name.find_last_of('.'));
                has_host = config.has(protocol + ".host");

                /// Whether any handler in the `impl` chain actually consumes the default session
                /// user, so that a `default_session_user` change does not restart listeners that
                /// ignore it: keeper-metrics-only `prometheus` listeners serve metrics without
                /// authentication (`KeeperPrometheusHandler-factory` only exposes `MetricsImpl`),
                /// and `prometheus` endpoints whose configured handler set consists only of
                /// fixed-user and non-authenticating handlers never consult the setting either.
                bool consumes_default_session_user = false;

                std::string conf_name = protocol;
                std::string prefix = protocol + ".";
                std::unordered_set<std::string> pset {conf_name};
                while (true)
                {
                    if (config.has(prefix + "type"))
                    {
                        std::string type = config.getString(prefix + "type");
                        if (type == "http")
                        {
                            is_http = true;
                            if (config.has(conf_name + ".handlers"))
                                handlers_key = config.getString(conf_name + ".handlers");
                            /// This reload may re-point the endpoint at a different `<handlers>` section
                            /// (through the endpoint's `handlers` reference or its `impl` chain), while
                            /// the running factory still serves the previously referenced section. Resolve
                            /// the old reference too: the consumer check must consider the old *or* the
                            /// new handler set, and the changed-configuration check below must notice the
                            /// reference switch itself — comparing only the new section on both sides of
                            /// the reload would miss a switch between two sections that are themselves
                            /// unchanged.
                            previous_handlers_key = resolveHTTPHandlersKey(previous_config, protocol).value_or(handlers_key);
                            /// `SQLDefinedHTTPHandlerFactory` is appended to every HTTP handler
                            /// factory. It caches the endpoint override, while the SQL handler
                            /// registry itself is read on every request, so an SQL-defined handler
                            /// can start consuming the setting after the listener was created even
                            /// when all XML handlers have fixed users.
                            consumes_default_session_user = true;
                            break;
                        }
                        if (type == "tcp" || type == "mysql" || type == "postgres")
                            consumes_default_session_user = true;
                        /// The running listener still serves the handler set of the *previous*
                        /// configuration (the shared `prometheus.handlers` section), so consider the
                        /// setting consumed when either the old or the new set consumes it. A change
                        /// of the section itself forces a restart below, like a change of an `http`
                        /// endpoint's `handlers` section.
                        if (type == "prometheus")
                        {
                            is_prometheus = true;
                            if (!server_settings[ServerSetting::prometheus_keeper_metrics_only])
                            {
                                is_non_keeper_prometheus = true;
                                consumes_default_session_user = consumes_default_session_user
                                    || prometheusHandlersConsumeDefaultSessionUser(previous_config)
                                    || prometheusHandlersConsumeDefaultSessionUser(config);
                            }
                        }
                    }

                    if (!config.has(prefix + "impl"))
                        break;

                    conf_name = "protocols." + config.getString(prefix + "impl");
                    prefix = conf_name + ".";

                    if (!pset.insert(conf_name).second)
                        throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Protocol '{}' configuration contains a loop on '{}'", protocol, conf_name);
                }

                /// The per-endpoint default session user is fixed in the protocol handler factory,
                /// so the endpoint must be restarted when its effective value changes. Compare the
                /// effective value (the one closest to the endpoint wins, as in
                /// `buildProtocolStackFromConfig`) rather than every node in the `impl` chain, so
                /// that editing a shadowed base module does not force a needless restart.
                default_session_user_changed = consumes_default_session_user
                    && getEffectiveDefaultSessionUser(previous_config, protocol)
                        != getEffectiveDefaultSessionUser(config, protocol);
            }
            else
            {
                /// NOTE: better to compare using getPortName() over using
                /// dynamic_cast<> since HTTPServer is also used for prometheus and
                /// internal replication communications.
                is_http = server->getPortName() == "http_port" || server->getPortName() == "https_port";
                is_prometheus = server->getPortName() == "prometheus.port";
                /// The standalone `prometheus.port` listener is built from the same shared
                /// `prometheus.handlers` section as a composable `type = prometheus` endpoint
                /// (`createPrometheusMainHandlerFactory` bakes it into the handler factory), and
                /// there is no other reconfiguration hook for it, so it needs the same restart
                /// check. In keeper-metrics-only mode the section is not served at all, and a
                /// change of the mode itself is handled by the check below.
                is_non_keeper_prometheus = is_prometheus && !server_settings[ServerSetting::prometheus_keeper_metrics_only];
            }

            if (!has_host)
            {
                const auto & server_listen_hosts =
#if USE_GRPC || USE_ARROWFLIGHT
                    (port_name == "grpc_port" || port_name == "arrowflight_port") ? grpc_listen_hosts :
#endif
                    listen_hosts;
                has_host = std::find(server_listen_hosts.begin(), server_listen_hosts.end(), server->getListenHost()) != server_listen_hosts.end();
            }
            bool has_port = !config.getString(port_name, "").empty();
            bool force_restart = false;
            if (is_http && previous_handlers_key != handlers_key)
            {
                force_restart = true;
                LOG_TRACE(log, "The handlers reference had been changed from <{}> to <{}>, will reload {}",
                    previous_handlers_key, handlers_key, server->getDescription());
            }
            else if (is_http && !isSameConfiguration(previous_config, config, handlers_key))
            {
                force_restart = true;
                LOG_TRACE(log, "<{}> had been changed, will reload {}", handlers_key, server->getDescription());
            }
            /// A non-keeper `prometheus` listener (a composable `type = prometheus` endpoint as well
            /// as the standalone `prometheus.port` one) serves the shared `prometheus.handlers`
            /// section, which is baked into its handler factory, so the listener must be restarted
            /// when the section changes (a fixed `user`, a route, or a handler type may have changed).
            if (is_non_keeper_prometheus && !isSameConfiguration(previous_config, config, "prometheus.handlers"))
            {
                force_restart = true;
                LOG_TRACE(log, "<prometheus.handlers> had been changed, will reload {}", server->getDescription());
            }
            /// `prometheus.keeper_metrics_only` selects the handler factory baked into a `prometheus`
            /// listener: `KeeperPrometheusHandler-factory` exposes the keeper metrics without
            /// authentication, while `PrometheusHandler-factory` also serves the authenticating
            /// time-series handlers of `prometheus.handlers`. A change of the mode itself must restart
            /// the listener in either direction — the port and the handler section may well stay the
            /// same, and neither of the checks above can notice the switched factory then.
            if (is_prometheus
                && previous_config.getBool("prometheus.keeper_metrics_only", false)
                    != server_settings[ServerSetting::prometheus_keeper_metrics_only])
            {
                force_restart = true;
                LOG_TRACE(log, "<prometheus.keeper_metrics_only> had been changed, will reload {}", server->getDescription());
            }
            /// The warehouse name is baked into the Iceberg REST catalog handler factory, so if
            /// the section changes, the listener must be restarted.
            if ((port_name == "iceberg_rest_catalog.port" || port_name == "iceberg_rest_catalog.secure_port")
                && !isSameConfiguration(previous_config, config, "iceberg_rest_catalog"))
            {
                force_restart = true;
                LOG_TRACE(log, "<iceberg_rest_catalog> had been changed, will reload {}", server->getDescription());
            }
            /// `asynchronous_metrics_key_values_mode` decides whether the keys of the key-value asynchronous
            /// metrics are written as Prometheus labels (`device="sda"`) or mangled into the metric name. A
            /// listener that exposes metrics is built with the constant labels and the form of its
            /// configuration, and neither is re-read while it runs, so it has to be rebuilt for the new form
            /// to be published with a label set that matches it. Only listeners that can actually serve the
            /// metrics protocol are rebuilt: a keeper-metrics-only `prometheus` listener exposes no
            /// asynchronous metrics at all, and an HTTP listener does so only through a rule with a
            /// `prometheus` handler type or through the default `/metrics` route - a change of the mode must
            /// not stop the HTTP interface of a server that only answers queries over it. The old and the new
            /// handler set are both consulted, because a rule may have just been added or removed.
            if (getAsynchronousMetricsKeyValuesMode(previous_config) != getAsynchronousMetricsKeyValuesMode(config)
                && (is_non_keeper_prometheus
                    || (is_http
                        && (httpHandlersCanExposePrometheusMetrics(previous_config, previous_handlers_key)
                            || httpHandlersCanExposePrometheusMetrics(config, handlers_key)))))
            {
                force_restart = true;
                LOG_TRACE(log, "<asynchronous_metrics_key_values_mode> had been changed, will reload {}", server->getDescription());
            }
            if (default_session_user_changed)
            {
                force_restart = true;
                LOG_TRACE(log, "<default_session_user> had been changed, will reload {}", server->getDescription());
            }

            if (!has_host || !has_port || config.getInt(server->getPortName()) != server->portNumber() || force_restart)
            {
                server->stop();
                LOG_INFO(log, "Stopped listening for {}", server->getDescription());
            }
        }
    }

    createServers(config, server_settings, listen_hosts, listen_try, server_pool, async_metrics, servers, /* start_servers= */ true);
    createInterserverServers(config, server_settings, interserver_listen_hosts, listen_try, server_pool, async_metrics, servers_to_start_before_tables, /* start_servers= */ true);

    std::erase_if(servers, std::bind_front(check_server, ""));
    std::erase_if(servers_to_start_before_tables, std::bind_front(check_server, ""));
}

}