{

LoggerPtr log = getLogger("Context");

/// For access of most of shared objects.

mutable ContextSharedMutex mutex;

/// Separate mutex for access of dictionaries. Separate mutex to avoid locks when server doing request to itself.

mutable std::mutex embedded_dictionaries_mutex;

mutable std::mutex external_dictionaries_mutex;

mutable std::mutex external_user_defined_executable_functions_mutex;

/// Separate mutex for storage policies. During server startup we may

/// initialize some important storages (system logs with MergeTree engine)

/// under context lock.

mutable std::mutex storage_policies_mutex;

/// Separate mutex for re-initialization of zookeeper session. This operation could take a long time and must not interfere with another operations.

mutable std::mutex zookeeper_mutex;

mutable zkutil::ZooKeeperPtr zookeeper TSA_GUARDED_BY(zookeeper_mutex); /// Client for ZooKeeper.

ConfigurationPtr zookeeper_config TSA_GUARDED_BY(zookeeper_mutex); /// Stores zookeeper configs

ConfigurationPtr sensitive_data_masker_config;

mutable std::mutex auxiliary_zookeepers_mutex;

mutable std::map<String, zkutil::ZooKeeperPtr> auxiliary_zookeepers TSA_GUARDED_BY(auxiliary_zookeepers_mutex); /// Map for auxiliary ZooKeeper clients.

ConfigurationPtr auxiliary_zookeepers_config TSA_GUARDED_BY(auxiliary_zookeepers_mutex); /// Stores auxiliary zookeepers configs

/// No lock required for interserver_io_host, interserver_io_port, interserver_scheme modified only during initialization

String interserver_io_host; /// The host name by which this server is available for other servers.

UInt16 interserver_io_port = 0; /// and port.

String interserver_scheme; /// http or https

MultiVersion<InterserverCredentials> interserver_io_credentials;

String path TSA_GUARDED_BY(mutex); /// Path to the data directory, with a slash at the end.

String flags_path TSA_GUARDED_BY(mutex); /// Path to the directory with some control flags for server maintenance.

String user_files_path TSA_GUARDED_BY(mutex); /// Path to the directory with user provided files, usable by 'file' table function.

String dictionaries_lib_path TSA_GUARDED_BY(mutex); /// Path to the directory with user provided binaries and libraries for external dictionaries.

String user_scripts_path TSA_GUARDED_BY(mutex); /// Path to the directory with user provided scripts.

String filesystem_caches_path TSA_GUARDED_BY(mutex); /// Path to the directory with filesystem caches.

String filesystem_cache_user TSA_GUARDED_BY(mutex);

ConfigurationPtr config TSA_GUARDED_BY(mutex); /// Global configuration settings.

String tmp_path TSA_GUARDED_BY(mutex); /// Path to the temporary files that occur when processing the request.

/// The default disk storing metadata files for databases: database metadata files and table metadata files.

/// For DBs which have `disk` setting in the create query, the table metadata files of these DBs are stored on that disk.

/// However, the DB metadata files are still stored on this `default_db_disk`. So the instance can load its DBs during starting up.

std::shared_ptr<IDisk> default_db_disk TSA_GUARDED_BY(mutex);

/// All temporary files that occur when processing the requests accounted here.

/// Child scopes for more fine-grained accounting are created per user/query/etc.

/// Initialized once during server startup.

TemporaryDataOnDiskScopePtr root_temp_data_on_disk TSA_GUARDED_BY(mutex);

/// TODO: remove, use only root_temp_data_on_disk

VolumePtr temporary_volume_legacy;

mutable OnceFlag async_loader_initialized;

mutable std::unique_ptr<AsyncLoader> async_loader; /// Thread pool for asynchronous initialization of arbitrary DAG of `LoadJob`s (used for tables loading)

mutable std::unique_ptr<EmbeddedDictionaries> embedded_dictionaries TSA_GUARDED_BY(embedded_dictionaries_mutex); /// Metrica's dictionaries. Have lazy initialization.

mutable std::unique_ptr<ExternalDictionariesLoader> external_dictionaries_loader TSA_GUARDED_BY(external_dictionaries_mutex);

ExternalLoaderXMLConfigRepository * external_dictionaries_config_repository TSA_GUARDED_BY(external_dictionaries_mutex) = nullptr;

scope_guard dictionaries_xmls TSA_GUARDED_BY(external_dictionaries_mutex);

mutable std::unique_ptr<ExternalUserDefinedExecutableFunctionsLoader> external_user_defined_executable_functions_loader TSA_GUARDED_BY(external_user_defined_executable_functions_mutex);

ExternalLoaderXMLConfigRepository * user_defined_executable_functions_config_repository TSA_GUARDED_BY(external_user_defined_executable_functions_mutex) = nullptr;

scope_guard user_defined_executable_functions_xmls TSA_GUARDED_BY(external_user_defined_executable_functions_mutex);

mutable OnceFlag user_defined_sql_objects_storage_initialized;

mutable std::unique_ptr<IUserDefinedSQLObjectsStorage> user_defined_sql_objects_storage;

mutable OnceFlag workload_entity_storage_initialized;

mutable std::unique_ptr<IWorkloadEntityStorage> workload_entity_storage;

mutable std::unique_ptr<WasmModuleManager> wasm_module_manager;

#if USE_NLP

mutable OnceFlag synonyms_extensions_initialized;

mutable std::optional<SynonymsExtensions> synonyms_extensions;

mutable OnceFlag lemmatizers_initialized;

mutable std::optional<Lemmatizers> lemmatizers;

#endif

mutable OnceFlag backups_worker_initialized;

std::optional<BackupsWorker> backups_worker;

/// No lock required for default_profile_name, system_profile_name, buffer_profile_name modified only during initialization

String default_profile_name; /// Default profile name used for default values.

String system_profile_name; /// Profile used by system processes

String background_profile_name; /// Profile used by background operations

String buffer_profile_name; /// Profile used by Buffer engine for flushing to the underlying

String merge_workload TSA_GUARDED_BY(mutex); /// Workload setting value that is used by all merges

String mutation_workload TSA_GUARDED_BY(mutex); /// Workload setting value that is used by all mutations

String license_file TSA_GUARDED_BY(mutex); /// BYOC license text

bool throw_on_unknown_workload TSA_GUARDED_BY(mutex) = false;

bool cpu_slot_preemption TSA_GUARDED_BY(mutex) = false;

UInt64 cpu_slot_quantum_ns TSA_GUARDED_BY(mutex) = 10'000'000;

UInt64 cpu_slot_preemption_timeout_ms TSA_GUARDED_BY(mutex) = 1000;

UInt64 concurrent_threads_soft_limit_num TSA_GUARDED_BY(mutex) = 0;

UInt64 concurrent_threads_soft_limit_ratio_to_cores TSA_GUARDED_BY(mutex) = 0;

String concurrent_threads_scheduler TSA_GUARDED_BY(mutex);

std::unique_ptr<AccessControl> access_control TSA_GUARDED_BY(mutex);

mutable OnceFlag resource_manager_initialized;

mutable ResourceManagerPtr resource_manager;

mutable UncompressedCachePtr uncompressed_cache TSA_GUARDED_BY(mutex); /// The cache of decompressed blocks.

mutable MarkCachePtr mark_cache TSA_GUARDED_BY(mutex); /// Cache of marks in compressed files.

mutable PrimaryIndexCachePtr primary_index_cache TSA_GUARDED_BY(mutex);

mutable SystemAllocatedMemoryHolderPtr untracked_memory_holder TSA_GUARDED_BY(mutex);

mutable OnceFlag load_marks_threadpool_initialized;

mutable std::unique_ptr<ThreadPool> load_marks_threadpool; /// Threadpool for loading marks cache.

mutable OnceFlag prefetch_threadpool_initialized;

mutable std::unique_ptr<ThreadPool> prefetch_threadpool; /// Threadpool for loading marks cache.

mutable std::unique_ptr<ThreadPool> iceberg_catalog_threadpool;

mutable OnceFlag iceberg_catalog_threadpool_initialized;

mutable OnceFlag build_vector_similarity_index_threadpool_initialized;

mutable std::unique_ptr<ThreadPool> build_vector_similarity_index_threadpool; /// Threadpool for vector-similarity index creation.

mutable UncompressedCachePtr index_uncompressed_cache TSA_GUARDED_BY(mutex); /// The cache of decompressed blocks for MergeTree indices.

mutable VectorSimilarityIndexCachePtr vector_similarity_index_cache TSA_GUARDED_BY(mutex); /// Cache of deserialized secondary index granules.

mutable TextIndexTokensCachePtr text_index_tokens_cache TSA_GUARDED_BY(mutex); /// Cache of deserialized text index tokens.

mutable TextIndexHeaderCachePtr text_index_header_cache TSA_GUARDED_BY(mutex); /// Cache of deserialized text index headers.

mutable TextIndexPostingsCachePtr text_index_postings_cache TSA_GUARDED_BY(mutex); /// Cache of deserialized text index posting lists.

mutable QueryConditionCachePtr query_condition_cache TSA_GUARDED_BY(mutex); /// Cache of matching marks for predicates

mutable QueryResultCachePtr query_result_cache TSA_GUARDED_BY(mutex); /// Cache of query results.

mutable MarkCachePtr index_mark_cache TSA_GUARDED_BY(mutex); /// Cache of marks in compressed files of MergeTree indices.

mutable MMappedFileCachePtr mmap_cache TSA_GUARDED_BY(mutex); /// Cache of mmapped files to avoid frequent open/map/unmap/close and to reuse from several threads.

#if USE_AVRO

mutable IcebergMetadataFilesCachePtr iceberg_metadata_files_cache TSA_GUARDED_BY(mutex); /// Cache of deserialized iceberg metadata files.

#endif

#if USE_PARQUET

mutable ParquetMetadataCachePtr parquet_metadata_cache TSA_GUARDED_BY(mutex); /// Cache of deserialized parquet metadata files.

#endif

AsynchronousMetrics * asynchronous_metrics TSA_GUARDED_BY(mutex) = nullptr; /// Points to asynchronous metrics

mutable PageCachePtr page_cache TSA_GUARDED_BY(mutex); /// Userspace page cache.

ProcessList process_list; /// Executing queries at the moment.

SessionTracker session_tracker;

GlobalOvercommitTracker global_overcommit_tracker;

MergeList merge_list; /// The list of executable merge (for (Replicated)?MergeTree)

MovesList moves_list; /// The list of executing moves (for (Replicated)?MergeTree)

ReplicatedFetchList replicated_fetch_list;

RefreshSet refresh_set; /// The list of active refreshes (for MaterializedView)

ConfigurationPtr users_config TSA_GUARDED_BY(mutex); /// Config with the users, profiles and quotas sections.

InterserverIOHandler interserver_io_handler; /// Handler for interserver communication.

OnceFlag buffer_flush_schedule_pool_initialized;

mutable BackgroundSchedulePoolPtr buffer_flush_schedule_pool; /// A thread pool that can do background flush for Buffer tables.

OnceFlag schedule_pool_initialized;

mutable BackgroundSchedulePoolPtr schedule_pool; /// A thread pool that can run different jobs in background (used in replicated tables)

OnceFlag distributed_schedule_pool_initialized;

mutable BackgroundSchedulePoolPtr distributed_schedule_pool; /// A thread pool that can run different jobs in background (used for distributed sends)

OnceFlag message_broker_schedule_pool_initialized;

mutable BackgroundSchedulePoolPtr message_broker_schedule_pool; /// A thread pool that can run different jobs in background (used for message brokers, like RabbitMQ and Kafka)

OnceFlag iceberg_schedule_pool_initialized;

mutable BackgroundSchedulePoolPtr iceberg_schedule_pool; /// A thread pool that runs background metadata refresh for all active Iceberg tables

mutable OnceFlag readers_initialized;

mutable std::unique_ptr<IAsynchronousReader> asynchronous_remote_fs_reader;

mutable std::unique_ptr<IAsynchronousReader> asynchronous_local_fs_reader;

mutable std::unique_ptr<IAsynchronousReader> synchronous_local_fs_reader;

mutable OnceFlag threadpool_writer_initialized;

mutable std::unique_ptr<ThreadPool> threadpool_writer;

#if USE_LIBURING

mutable OnceFlag io_uring_reader_initialized;

mutable std::unique_ptr<IOUringReader> io_uring_reader;

#endif

mutable ThrottlerPtr replicated_fetches_throttler; /// A server-wide throttler for replicated fetches

mutable ThrottlerPtr replicated_sends_throttler; /// A server-wide throttler for replicated sends

mutable ThrottlerPtr remote_read_throttler; /// A server-wide throttler for remote IO reads

mutable ThrottlerPtr remote_write_throttler; /// A server-wide throttler for remote IO writes

mutable ThrottlerPtr local_read_throttler; /// A server-wide throttler for local IO reads

mutable ThrottlerPtr local_write_throttler; /// A server-wide throttler for local IO writes

mutable ThrottlerPtr backups_server_throttler; /// A server-wide throttler for BACKUPs

mutable ThrottlerPtr mutations_throttler; /// A server-wide throttler for mutations

mutable ThrottlerPtr merges_throttler; /// A server-wide throttler for merges

mutable ThrottlerPtr distributed_cache_read_throttler; /// A server-wide throttler for distributed cache read

mutable ThrottlerPtr distributed_cache_write_throttler; /// A server-wide throttler for distributed cache write

MultiVersion<Macros> macros; /// Substitutions extracted from config.

std::unique_ptr<DDLWorker> ddl_worker TSA_GUARDED_BY(mutex); /// Process ddl commands from zk.

LoadTaskPtr ddl_worker_startup_task; /// To postpone `ddl_worker->startup()` after all tables startup

/// Rules for selecting the compression settings, depending on the size of the part.

mutable std::unique_ptr<CompressionCodecSelector> compression_codec_selector TSA_GUARDED_BY(mutex);

/// Storage disk chooser for MergeTree engines

mutable std::shared_ptr<const DiskSelector> merge_tree_disk_selector TSA_GUARDED_BY(storage_policies_mutex);

/// Storage policy chooser for MergeTree engines

mutable std::shared_ptr<const StoragePolicySelector> merge_tree_storage_policy_selector TSA_GUARDED_BY(storage_policies_mutex);

ServerSettings server_settings;

std::optional<MergeTreeSettings> merge_tree_settings TSA_GUARDED_BY(mutex); /// Settings of MergeTree* engines.

std::optional<MergeTreeSettings> replicated_merge_tree_settings TSA_GUARDED_BY(mutex); /// Settings of ReplicatedMergeTree* engines.

std::optional<DatabaseReplicatedSettings> database_replicated_settings TSA_GUARDED_BY(mutex); /// Settings of DatabaseReplicated engine.

std::optional<DistributedSettings> distributed_settings TSA_GUARDED_BY(mutex);

std::atomic_size_t max_table_size_to_drop = 50000000000lu; /// Protects MergeTree tables from accidental DROP (50GB by default)

std::atomic_size_t max_partition_size_to_drop = 50000000000lu; /// Protects MergeTree partitions from accidental DROP (50GB by default)

/// No lock required for format_schema_path modified only during initialization

std::atomic_size_t max_database_num_to_warn = 1000lu;

std::atomic_size_t max_named_collection_num_to_warn = 1000lu;

std::atomic_size_t max_table_num_to_warn = 5000lu;

std::atomic_size_t max_view_num_to_warn = 10000lu;

std::atomic_size_t max_dictionary_num_to_warn = 1000lu;

std::atomic_size_t max_part_num_to_warn = 100000lu;

// these variables are used in inserting warning message into system.warning table based on asynchronous metrics

size_t max_pending_mutations_to_warn = 500lu;

size_t max_pending_mutations_execution_time_to_warn = 86400lu;

/// Only for system.server_settings, actually value stored in reloader itself

std::atomic_size_t config_reload_interval_ms = ConfigReloader::DEFAULT_RELOAD_INTERVAL.count();

double min_os_cpu_wait_time_ratio_to_drop_connection = 15.0;

double max_os_cpu_wait_time_ratio_to_drop_connection = 30.0;

String format_schema_path; /// Path to a directory that contains schema files used by input formats.

String google_protos_path; /// Path to a directory that contains the proto files for the well-known Protobuf types.

mutable OnceFlag action_locks_manager_initialized;

ActionLocksManagerPtr action_locks_manager; /// Set of storages' action lockers

OnceFlag system_logs_initialized;

std::unique_ptr<SystemLogs> system_logs TSA_GUARDED_BY(mutex); /// Used to log queries and operations on parts

mutable std::mutex dashboard_mutex;

std::optional<Context::Dashboards> dashboards;

mutable SharedMutex users_to_ignore_early_memory_limit_check_mutex;

std::string users_to_ignore_early_memory_limit_check_source TSA_GUARDED_BY(users_to_ignore_early_memory_limit_check_mutex);

std::shared_ptr<std::unordered_set<std::string>> users_to_ignore_early_memory_limit_check TSA_GUARDED_BY(users_to_ignore_early_memory_limit_check_mutex);

std::optional<S3SettingsByEndpoint> storage_s3_settings TSA_GUARDED_BY(mutex); /// Settings of S3 storage

std::optional<AzureSettingsByEndpoint> storage_azure_settings TSA_GUARDED_BY(mutex); /// Settings of AzureBlobStorage

std::unordered_map<Context::WarningType, PreformattedMessage> warnings TSA_GUARDED_BY(mutex); /// Store warning messages about server.

/// Background executors for *MergeTree tables

/// Has background executors for MergeTree tables been initialized?

mutable ContextSharedMutex background_executors_mutex;

bool are_background_executors_initialized TSA_GUARDED_BY(background_executors_mutex) = false;

MergeMutateBackgroundExecutorPtr merge_mutate_executor TSA_GUARDED_BY(background_executors_mutex);

OrdinaryBackgroundExecutorPtr moves_executor TSA_GUARDED_BY(background_executors_mutex);

OrdinaryBackgroundExecutorPtr fetch_executor TSA_GUARDED_BY(background_executors_mutex);

OrdinaryBackgroundExecutorPtr common_executor TSA_GUARDED_BY(background_executors_mutex);

RemoteHostFilter remote_host_filter; /// Allowed URL from config.xml

HTTPHeaderFilter http_header_filter; /// Forbidden HTTP headers from config.xml

/// No lock required for trace_collector modified only during initialization

std::optional<TraceCollector> trace_collector; /// Thread collecting traces from threads executing queries

/// Clusters for distributed tables

/// Initialized on demand (on distributed storages initialization) since Settings should be initialized

mutable std::mutex clusters_mutex; /// Guards clusters, clusters_config and cluster_discovery

std::shared_ptr<Clusters> clusters TSA_GUARDED_BY(clusters_mutex);

ConfigurationPtr clusters_config TSA_GUARDED_BY(clusters_mutex); /// Stores updated configs

std::unique_ptr<ClusterDiscovery> cluster_discovery TSA_GUARDED_BY(clusters_mutex);

size_t clusters_version TSA_GUARDED_BY(clusters_mutex) = 0;

/// No lock required for async_insert_queue modified only during initialization

std::shared_ptr<AsynchronousInsertQueue> async_insert_queue;

std::map<String, UInt16> server_ports;

std::atomic<bool> shutdown_called = false;

Stopwatch uptime_watch TSA_GUARDED_BY(mutex);

/// No lock required for application_type modified only during initialization

Context::ApplicationType application_type = Context::ApplicationType::SERVER;

/// No lock required for config_reload_callback, start_servers_callback, stop_servers_callback modified only during initialization

Context::ConfigReloadCallback config_reload_callback;

Context::StartStopServersCallback start_servers_callback;

Context::StartStopServersCallback stop_servers_callback;

bool is_server_completely_started TSA_GUARDED_BY(mutex) = false;

#if USE_NURAFT

mutable std::shared_ptr<KeeperDispatcher> keeper_dispatcher;

#endif

ContextSharedPart()

: access_control(std::make_unique<AccessControl>()), global_overcommit_tracker(&process_list), macros(std::make_unique<Macros>())

{

/// TODO: make it singleton (?)

static std::atomic<size_t> num_calls{0};

if (++num_calls > 1)

{

std::cerr << "Attempting to create multiple ContextShared instances. Stack trace:\n" << StackTrace().toString();

std::cerr.flush();

std::terminate();

}

}

~ContextSharedPart()

{

/// Shutdown must be called first to stop all background tasks (like loadOutdatedDataParts)

/// that may be using the thread pool readers. Otherwise there is a data race between

/// background tasks calling getThreadPoolReader() and the destructor resetting the readers.

/// See https://github.com/ClickHouse/ClickHouse/issues/62143

try

{

shutdown();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

#if USE_NURAFT

if (keeper_dispatcher)

{

try

{

keeper_dispatcher->shutdown();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

#endif

/// Wait for thread pool for background reads and writes,

/// since it may use per-user MemoryTracker which will be destroyed here.

if (asynchronous_remote_fs_reader)

{

try

{

LOG_DEBUG(log, "Destructing remote fs threadpool reader");

asynchronous_remote_fs_reader->wait();

asynchronous_remote_fs_reader.reset();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

if (asynchronous_local_fs_reader)

{

try

{

LOG_DEBUG(log, "Destructing local fs threadpool reader");

asynchronous_local_fs_reader->wait();

asynchronous_local_fs_reader.reset();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

if (synchronous_local_fs_reader)

{

try

{

LOG_DEBUG(log, "Destructing local fs threadpool reader");

synchronous_local_fs_reader->wait();

synchronous_local_fs_reader.reset();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

if (threadpool_writer)

{

try

{

LOG_DEBUG(log, "Destructing threadpool writer");

threadpool_writer->wait();

threadpool_writer.reset();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

if (load_marks_threadpool)

{

try

{

LOG_DEBUG(log, "Destructing marks loader");

load_marks_threadpool->wait();

load_marks_threadpool.reset();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

if (prefetch_threadpool)

{

try

{

LOG_DEBUG(log, "Destructing prefetch threadpool");

prefetch_threadpool->wait();

prefetch_threadpool.reset();

}

catch (...)

{

tryLogCurrentException(__PRETTY_FUNCTION__);

}

}

}

void setConfig(const ConfigurationPtr & config_value)

{

if (!config_value)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Set nullptr config is invalid");

std::lock_guard lock(mutex);

config = config_value;

access_control->setExternalAuthenticatorsConfig(*config_value);

}

const Poco::Util::AbstractConfiguration & getConfigRefWithLock(const std::lock_guard<ContextSharedMutex> &) const TSA_REQUIRES(this->mutex)

{

return config ? *config : Poco::Util::Application::instance().config();

}

const Poco::Util::AbstractConfiguration & getConfigRef() const

{

SharedLockGuard lock(mutex);

return config ? *config : Poco::Util::Application::instance().config();

}

/** Perform a complex job of destroying objects in advance.

void shutdown() TSA_NO_THREAD_SAFETY_ANALYSIS

{

bool is_shutdown_called = shutdown_called.exchange(true);

if (is_shutdown_called)

return;

/// Need to flush the async insert queue before shutting down the database catalog

std::shared_ptr<AsynchronousInsertQueue> delete_async_insert_queue;

{

std::lock_guard lock(mutex);

delete_async_insert_queue = std::move(async_insert_queue);

}

if (delete_async_insert_queue)

delete_async_insert_queue->flushAndShutdown();

/// Stop periodic reloading of the configuration files.

/// This must be done first because otherwise the reloading may pass a changed config

/// to some destroyed parts of ContextSharedPart.

/// We need to make sure no dictionary queries are concurrent with the further shutdown logic

/// because it sets various fields to null and those fields are unconditionally dereferenced by queries.

/// So we need to:

/// (1) Disable future dictionary queries.

/// (2) Make sure all the currently running dictionary queries are killed.

/// (3) Join the dictionary queries' threads.

SHUTDOWN(log, "dictionaries loader", external_dictionaries_loader, enablePeriodicUpdates(false));

process_list.killAllQueries();

SHUTDOWN(log, "dictionaries loader threads", external_dictionaries_loader, joinLoadingThreads());

SHUTDOWN(log, "UDFs loader", external_user_defined_executable_functions_loader, enablePeriodicUpdates(false));

SHUTDOWN(log, "another UDFs storage", user_defined_sql_objects_storage, stopWatching());

LOG_TRACE(log, "Shutting down named sessions");

Session::shutdownNamedSessions();

/// Stop watching DDL queue in ZooKeeper and wait until currently executed tasks finish.

/// This must be done before closing ZooKeeper connection (because DDLWorker can call Context::getZooKeeper() and resurrect it),

/// and before shutting down BackupsWorker (because DDLWorker can start an internal backup or restore).

SHUTDOWN(log, "ddl worker", ddl_worker, shutdown());

/// Waiting for current backups/restores to be finished. This must be done before shutting down DatabaseCatalog.

SHUTDOWN(log, "backups worker", backups_worker, shutdown());

LOG_TRACE(log, "Shutting down object storage queue streaming");

StreamingStorageRegistry::instance().shutdown();

/// Stop all MergeTree background executors before shutting down databases.

/// This ensures no background tasks (merges, mutations, moves, part cleanup)

/// are running when storage objects are shut down or destroyed.

/// Without this, a background task could be accessing a storage's data_parts_indexes

/// while DatabaseCatalog::shutdown is destroying that storage, causing a SIGBUS.

/// See https://github.com/ClickHouse/ClickHouse/issues/85433

SHUTDOWN(log, "merges executor", merge_mutate_executor, wait());

SHUTDOWN(log, "fetches executor", fetch_executor, wait());

SHUTDOWN(log, "moves executor", moves_executor, wait());

SHUTDOWN(log, "common executor", common_executor, wait());

/// Deactivate FileCache background threads before shutting down the database catalog.

/// DatabaseCatalog::shutdown() can throw (e.g. ZooKeeper timeout in replicated DDL worker).

/// If it throws, deactivateBackgroundOperations() would be skipped, leaving FileCache

/// download/cleanup threads stuck on GlobalThreadPool, which causes

/// GlobalThreadPool::shutdown() to deadlock.

LOG_TRACE(log, "Shutting down caches");

for (const auto & cache_data : FileCacheFactory::instance().getUniqueInstances())

cache_data->cache->deactivateBackgroundOperations();

LOG_TRACE(log, "Shutting down database catalog");

DatabaseCatalog::shutdown([this]()

{

SHUTDOWN(log, "system logs", TSA_SUPPRESS_WARNING_FOR_READ(system_logs), flushAndShutdown());

});

FileCacheFactory::instance().clear();

NamedCollectionFactory::instance().shutdown();

delete_async_insert_queue.reset();

TransactionLog::shutdownIfAny();

// Workload entity storage must be destructed when no queries or merges are running because PipelineExecutor may access it.

SHUTDOWN(log, "workload entity storage", workload_entity_storage, stopWatching());

std::unique_ptr<SystemLogs> delete_system_logs;

std::unique_ptr<EmbeddedDictionaries> delete_embedded_dictionaries;

std::unique_ptr<ExternalDictionariesLoader> delete_external_dictionaries_loader;

std::unique_ptr<ExternalUserDefinedExecutableFunctionsLoader> delete_external_user_defined_executable_functions_loader;

std::unique_ptr<IUserDefinedSQLObjectsStorage> delete_user_defined_sql_objects_storage;

std::unique_ptr<IWorkloadEntityStorage> delete_workload_entity_storage;

std::unique_ptr<DDLWorker> delete_ddl_worker;

BackgroundSchedulePoolPtr delete_buffer_flush_schedule_pool;

BackgroundSchedulePoolPtr delete_schedule_pool;

BackgroundSchedulePoolPtr delete_distributed_schedule_pool;

BackgroundSchedulePoolPtr delete_message_broker_schedule_pool;

BackgroundSchedulePoolPtr delete_iceberg_schedule_pool;

std::unique_ptr<AccessControl> delete_access_control;

scope_guard delete_dictionaries_xmls;

scope_guard delete_user_defined_executable_functions_xmls;

{

std::lock_guard lock(clusters_mutex);

if (cluster_discovery)

{

LOG_TRACE(log, "Shutting down ClusterDiscovery");

/// Reset cluster_discovery if any.

/// Some classes (such as ZooKeeper, ReplicatedAccessStorage) will finalize the keeper session while deconstructing,

/// which will trigger the callback and make ClusterDiscovery reconnect to keeper again (unnecessary).

cluster_discovery.reset();

}

}

{

// Disk selector might not be initialized if there was some error during

// its initialization. Don't try to initialize it again on shutdown.

if (merge_tree_disk_selector)

{

for (const auto & [disk_name, disk] : merge_tree_disk_selector->getDisksMap())

{

LOG_INFO(log, "Shutdown disk {}", disk_name);

disk->shutdown();

}

}

/// Special volumes might also use disks that require shutdown.

if (temporary_volume_legacy)

{