Introduce SettingsProfile as a new access entity type.

dbms/src/Access/AccessControlManager.cpp

@@ -8,6 +8,7 @@
 #include <Access/RowPolicyCache.h>
 #include <Access/QuotaCache.h>
 #include <Access/QuotaUsageInfo.h>
+#include <Access/SettingsProfilesCache.h>
 #include <Core/Settings.h>
 #include <Poco/ExpireCache.h>
 #include <mutex>
@@ -36,7 +37,13 @@ class AccessControlManager::ContextAccessCache
 public:
     explicit ContextAccessCache(const AccessControlManager & manager_) : manager(manager_) {}
 
-    std::shared_ptr<const ContextAccess> getContextAccess(const UUID & user_id, const std::vector<UUID> & current_roles, bool use_default_roles, const Settings & settings, const String & current_database, const ClientInfo & client_info)
+    std::shared_ptr<const ContextAccess> getContextAccess(
+        const UUID & user_id,
+        const std::vector<UUID> & current_roles,
+        bool use_default_roles,
+        const Settings & settings,
+        const String & current_database,
+        const ClientInfo & client_info)
     {
         ContextAccess::Params params;
         params.user_id = user_id;
@@ -72,7 +79,8 @@ AccessControlManager::AccessControlManager()
       context_access_cache(std::make_unique<ContextAccessCache>(*this)),
       role_cache(std::make_unique<RoleCache>(*this)),
       row_policy_cache(std::make_unique<RowPolicyCache>(*this)),
-      quota_cache(std::make_unique<QuotaCache>(*this))
+      quota_cache(std::make_unique<QuotaCache>(*this)),
+      settings_profiles_cache(std::make_unique<SettingsProfilesCache>(*this))
 {
 }
 
@@ -94,6 +102,12 @@ void AccessControlManager::setUsersConfig(const Poco::Util::AbstractConfiguratio
 }
 
 
+void AccessControlManager::setDefaultProfileName(const String & default_profile_name)
+{
+    settings_profiles_cache->setDefaultProfileName(default_profile_name);
+}
+
+
 std::shared_ptr<const ContextAccess> AccessControlManager::getContextAccess(
     const UUID & user_id,
     const std::vector<UUID> & current_roles,
@@ -132,4 +146,19 @@ std::vector<QuotaUsageInfo> AccessControlManager::getQuotaUsageInfo() const
     return quota_cache->getUsageInfo();
 }
 
+
+std::shared_ptr<const EnabledSettings> AccessControlManager::getEnabledSettings(
+    const UUID & user_id,
+    const SettingsProfileElements & settings_from_user,
+    const std::vector<UUID> & enabled_roles,
+    const SettingsProfileElements & settings_from_enabled_roles) const
+{
+    return settings_profiles_cache->getEnabledSettings(user_id, settings_from_user, enabled_roles, settings_from_enabled_roles);
+}
+
+std::shared_ptr<const SettingsChanges> AccessControlManager::getProfileSettings(const String & profile_name) const
+{
+    return settings_profiles_cache->getProfileSettings(profile_name);
+}
+
 }

dbms/src/Access/AccessControlManager.h

@@ -29,6 +29,11 @@ class RowPolicyCache;
 class EnabledQuota;
 class QuotaCache;
 struct QuotaUsageInfo;
+struct SettingsProfile;
+using SettingsProfilePtr = std::shared_ptr<const SettingsProfile>;
+class EnabledSettings;
+class SettingsProfilesCache;
+class SettingsProfileElements;
 class ClientInfo;
 struct Settings;
 
@@ -42,6 +47,7 @@ public:
 
     void setLocalDirectory(const String & directory);
     void setUsersConfig(const Poco::Util::AbstractConfiguration & users_config);
+    void setDefaultProfileName(const String & default_profile_name);
 
     std::shared_ptr<const ContextAccess> getContextAccess(
         const UUID & user_id,
@@ -68,12 +74,20 @@ public:
 
     std::vector<QuotaUsageInfo> getQuotaUsageInfo() const;
 
+    std::shared_ptr<const EnabledSettings> getEnabledSettings(const UUID & user_id,
+                                                              const SettingsProfileElements & settings_from_user,
+                                                              const std::vector<UUID> & enabled_roles,
+                                                              const SettingsProfileElements & settings_from_enabled_roles) const;
+
+    std::shared_ptr<const SettingsChanges> getProfileSettings(const String & profile_name) const;
+
 private:
     class ContextAccessCache;
     std::unique_ptr<ContextAccessCache> context_access_cache;
     std::unique_ptr<RoleCache> role_cache;
     std::unique_ptr<RowPolicyCache> row_policy_cache;
     std::unique_ptr<QuotaCache> quota_cache;
+    std::unique_ptr<SettingsProfilesCache> settings_profiles_cache;
 };
 
 }

dbms/src/Access/ContextAccess.cpp

@@ -5,6 +5,7 @@
 #include <Access/EnabledQuota.h>
 #include <Access/User.h>
 #include <Access/EnabledRolesInfo.h>
+#include <Access/EnabledSettings.h>
 #include <Interpreters/DatabaseCatalog.h>
 #include <Common/Exception.h>
 #include <Common/quoteString.h>
@@ -123,6 +124,7 @@ void ContextAccess::setUser(const UserPtr & user_) const
         roles_with_admin_option = nullptr;
         enabled_row_policies = nullptr;
         enabled_quota = nullptr;
+        enabled_settings = nullptr;
         return;
     }
 
@@ -172,6 +174,7 @@ void ContextAccess::setRolesInfo(const std::shared_ptr<const EnabledRolesInfo> &
     boost::range::fill(result_access, nullptr /* need recalculate */);
     enabled_row_policies = manager->getEnabledRowPolicies(*params.user_id, roles_info->enabled_roles);
     enabled_quota = manager->getEnabledQuota(*params.user_id, user_name, roles_info->enabled_roles, params.address, params.quota_key);
+    enabled_settings = manager->getEnabledSettings(*params.user_id, user->settings, roles_info->enabled_roles, roles_info->settings_from_enabled_roles);
 }
 
 
@@ -532,4 +535,18 @@ std::shared_ptr<const ContextAccess> ContextAccess::getFullAccess()
     return res;
 }
 
+
+std::shared_ptr<const Settings> ContextAccess::getDefaultSettings() const
+{
+    std::lock_guard lock{mutex};
+    return enabled_settings->getSettings();
+}
+
+
+std::shared_ptr<const SettingsConstraints> ContextAccess::getSettingsConstraints() const
+{
+    std::lock_guard lock{mutex};
+    return enabled_settings->getConstraints();
+}
+
 }

dbms/src/Access/ContextAccess.h

@@ -21,7 +21,9 @@ struct EnabledRolesInfo;
 class EnabledRoles;
 class EnabledRowPolicies;
 class EnabledQuota;
+class EnabledSettings;
 struct Settings;
+class SettingsConstraints;
 class AccessControlManager;
 class IAST;
 using ASTPtr = std::shared_ptr<IAST>;
@@ -69,6 +71,8 @@ public:
     std::shared_ptr<const EnabledRowPolicies> getRowPolicies() const;
     ASTPtr getRowPolicyCondition(const String & database, const String & table_name, RowPolicy::ConditionType index, const ASTPtr & extra_condition = nullptr) const;
     std::shared_ptr<const EnabledQuota> getQuota() const;
+    std::shared_ptr<const Settings> getDefaultSettings() const;
+    std::shared_ptr<const SettingsConstraints> getSettingsConstraints() const;
 
     /// Checks if a specified access is granted, and throws an exception if not.
     /// Empty database means the current database.
@@ -124,6 +128,7 @@ private:
 
     void setUser(const UserPtr & user_) const;
     void setRolesInfo(const std::shared_ptr<const EnabledRolesInfo> & roles_info_) const;
+    void setSettingsAndConstraints() const;
 
     template <int mode, bool grant_option, typename... Args>
     bool checkAccessImpl(Poco::Logger * log_, const AccessFlags & flags, const Args &... args) const;
@@ -150,6 +155,7 @@ private:
     mutable boost::atomic_shared_ptr<const AccessRights> result_access[7];
     mutable std::shared_ptr<const EnabledRowPolicies> enabled_row_policies;
     mutable std::shared_ptr<const EnabledQuota> enabled_quota;
+    mutable std::shared_ptr<const EnabledSettings> enabled_settings;
     mutable std::mutex mutex;
 };
 

dbms/src/Access/EnabledRolesInfo.cpp

@@ -28,7 +28,8 @@ bool operator==(const EnabledRolesInfo & lhs, const EnabledRolesInfo & rhs)
 {
     return (lhs.current_roles == rhs.current_roles) && (lhs.enabled_roles == rhs.enabled_roles)
         && (lhs.enabled_roles_with_admin_option == rhs.enabled_roles_with_admin_option) && (lhs.names_of_roles == rhs.names_of_roles)
-        && (lhs.access == rhs.access) && (lhs.access_with_grant_option == rhs.access_with_grant_option);
+        && (lhs.access == rhs.access) && (lhs.access_with_grant_option == rhs.access_with_grant_option)
+        && (lhs.settings_from_enabled_roles == rhs.settings_from_enabled_roles);
 }
 
 }

dbms/src/Access/EnabledRolesInfo.h

 #pragma once
 
 #include <Access/AccessRights.h>
+#include <Access/SettingsProfileElement.h>
 #include <Core/UUID.h>
 #include <unordered_map>
 #include <vector>
@@ -18,6 +19,7 @@ struct EnabledRolesInfo
     std::unordered_map<UUID, String> names_of_roles;
     AccessRights access;
     AccessRights access_with_grant_option;
+    SettingsProfileElements settings_from_enabled_roles;
 
     Strings getCurrentRolesNames() const;
     Strings getEnabledRolesNames() const;

dbms/src/Access/EnabledSettings.cpp

+#include <Access/EnabledSettings.h>
+
+
+namespace DB
+{
+
+EnabledSettings::EnabledSettings(const Params & params_) : params(params_)
+{
+}
+
+EnabledSettings::~EnabledSettings() = default;
+
+
+std::shared_ptr<const Settings> EnabledSettings::getSettings() const
+{
+    std::lock_guard lock{mutex};
+    return settings;
+}
+
+
+std::shared_ptr<const SettingsConstraints> EnabledSettings::getConstraints() const
+{
+    std::lock_guard lock{mutex};
+    return constraints;
+}
+
+
+void EnabledSettings::setSettingsAndConstraints(
+    const std::shared_ptr<const Settings> & settings_, const std::shared_ptr<const SettingsConstraints> & constraints_)
+{
+    std::lock_guard lock{mutex};
+    settings = settings_;
+    constraints = constraints_;
+}
+
+}

dbms/src/Access/EnabledSettings.h

+#pragma once
+
+#include <Core/Types.h>
+#include <Core/UUID.h>
+#include <Common/SettingsChanges.h>
+#include <Access/SettingsConstraints.h>
+#include <Access/SettingsProfileElement.h>
+#include <mutex>
+
+
+namespace DB
+{
+/// Watches settings profiles for a specific user and roles.
+class EnabledSettings
+{
+public:
+    struct Params
+    {
+        UUID user_id;
+        std::vector<UUID> enabled_roles;
+        SettingsProfileElements settings_from_enabled_roles;
+        SettingsProfileElements settings_from_user;
+
+        auto toTuple() const { return std::tie(user_id, enabled_roles, settings_from_enabled_roles, settings_from_user); }
+        friend bool operator ==(const Params & lhs, const Params & rhs) { return lhs.toTuple() == rhs.toTuple(); }
+        friend bool operator !=(const Params & lhs, const Params & rhs) { return !(lhs == rhs); }
+        friend bool operator <(const Params & lhs, const Params & rhs) { return lhs.toTuple() < rhs.toTuple(); }
+        friend bool operator >(const Params & lhs, const Params & rhs) { return rhs < lhs; }
+        friend bool operator <=(const Params & lhs, const Params & rhs) { return !(rhs < lhs); }
+        friend bool operator >=(const Params & lhs, const Params & rhs) { return !(lhs < rhs); }
+    };
+
+    ~EnabledSettings();
+
+    /// Returns the default settings come from settings profiles defined for the user
+    /// and the roles passed in the constructor.
+    std::shared_ptr<const Settings> getSettings() const;
+
+    /// Returns the constraints come from settings profiles defined for the user
+    /// and the roles passed in the constructor.
+    std::shared_ptr<const SettingsConstraints> getConstraints() const;
+
+private:
+    friend class SettingsProfilesCache;
+    EnabledSettings(const Params & params_);
+
+    void setSettingsAndConstraints(
+        const std::shared_ptr<const Settings> & settings_, const std::shared_ptr<const SettingsConstraints> & constraints_);
+
+    const Params params;
+    SettingsProfileElements settings_from_enabled;
+    std::shared_ptr<const Settings> settings;
+    std::shared_ptr<const SettingsConstraints> constraints;
+    mutable std::mutex mutex;
+};
+}

dbms/src/Access/Role.cpp

@@ -10,7 +10,8 @@ bool Role::equal(const IAccessEntity & other) const
         return false;
     const auto & other_role = typeid_cast<const Role &>(other);
     return (access == other_role.access) && (access_with_grant_option == other_role.access_with_grant_option)
-        && (granted_roles == other_role.granted_roles) && (granted_roles_with_admin_option == other_role.granted_roles_with_admin_option);
+        && (granted_roles == other_role.granted_roles) && (granted_roles_with_admin_option == other_role.granted_roles_with_admin_option)
+        && (settings == other_role.settings);
 }
 
 }

dbms/src/Access/Role.h

@@ -2,6 +2,7 @@
 
 #include <Access/IAccessEntity.h>
 #include <Access/AccessRights.h>
+#include <Access/SettingsProfileElement.h>
 #include <Core/UUID.h>
 #include <boost/container/flat_set.hpp>
 
@@ -15,6 +16,7 @@ struct Role : public IAccessEntity
     AccessRights access_with_grant_option;
     boost::container::flat_set<UUID> granted_roles;
     boost::container::flat_set<UUID> granted_roles_with_admin_option;
+    SettingsProfileElements settings;
 
     bool equal(const IAccessEntity & other) const override;
     std::shared_ptr<IAccessEntity> clone() const override { return cloneImpl<Role>(); }

dbms/src/Access/RoleCache.cpp

@@ -61,6 +61,7 @@ namespace
             new_info->names_of_roles[role_id] = role->getName();
             new_info->access.merge(role->access);
             new_info->access_with_grant_option.merge(role->access_with_grant_option);
+            new_info->settings_from_enabled_roles.merge(role->settings);
         }
         return new_info;
     }

dbms/src/Access/SettingsProfile.cpp

+#include <Access/SettingsProfile.h>
+
+
+namespace DB
+{
+bool SettingsProfile::equal(const IAccessEntity & other) const
+{
+    if (!IAccessEntity::equal(other))
+        return false;
+    const auto & other_profile = typeid_cast<const SettingsProfile &>(other);
+    return (elements == other_profile.elements) && (to_roles == other_profile.to_roles);
+}
+}

dbms/src/Access/SettingsProfile.h

+#pragma once
+
+#include <Access/IAccessEntity.h>
+#include <Access/ExtendedRoleSet.h>
+#include <Access/SettingsProfileElement.h>
+
+
+namespace DB
+{
+/// Represents a settings profile created by command
+/// CREATE SETTINGS PROFILE name SETTINGS x=value MIN=min MAX=max READONLY... TO roles
+struct SettingsProfile : public IAccessEntity
+{
+    SettingsProfileElements elements;
+
+    /// Which roles or users should use this settings profile.
+    ExtendedRoleSet to_roles;
+
+    bool equal(const IAccessEntity & other) const override;
+    std::shared_ptr<IAccessEntity> clone() const override { return cloneImpl<SettingsProfile>(); }
+};
+
+using SettingsProfilePtr = std::shared_ptr<const SettingsProfile>;
+}

dbms/src/Access/SettingsProfileElement.cpp

+#include <Access/SettingsProfileElement.h>
+#include <Access/SettingsConstraints.h>
+#include <Core/Settings.h>
+
+
+namespace DB
+{
+void SettingsProfileElements::merge(const SettingsProfileElements & other)
+{
+    insert(end(), other.begin(), other.end());
+}
+
+
+Settings SettingsProfileElements::toSettings() const
+{
+    Settings res;
+    for (const auto & elem : *this)
+    {
+        if (!elem.name.empty() && !elem.value.isNull())
+            res.set(elem.name, elem.value);
+    }
+    return res;
+}
+
+SettingsChanges SettingsProfileElements::toSettingsChanges() const
+{
+    SettingsChanges res;
+    for (const auto & elem : *this)
+    {
+        if (!elem.name.empty() && !elem.value.isNull())
+            res.push_back({elem.name, elem.value});
+    }
+    return res;
+}
+
+SettingsConstraints SettingsProfileElements::toSettingsConstraints() const
+{
+    SettingsConstraints res;
+    for (const auto & elem : *this)
+    {
+        if (!elem.name.empty())
+        {
+            if (!elem.min_value.isNull())
+                res.setMinValue(elem.name, elem.min_value);
+            if (!elem.max_value.isNull())
+                res.setMaxValue(elem.name, elem.max_value);
+            if (elem.readonly)
+                res.setReadOnly(elem.name, *elem.readonly);
+        }
+    }
+    return res;
+}
+
+}

dbms/src/Access/SettingsProfileElement.h

+#pragma once
+
+#include <Core/Field.h>
+#include <Core/UUID.h>
+#include <optional>
+#include <vector>
+
+
+namespace DB
+{
+struct Settings;
+struct SettingChange;
+using SettingsChanges = std::vector<SettingChange>;
+class SettingsConstraints;
+
+
+struct SettingsProfileElement
+{
+    std::optional<UUID> parent_profile;
+    String name;
+    Field value;
+    Field min_value;
+    Field max_value;
+    std::optional<bool> readonly;
+
+    auto toTuple() const { return std::tie(parent_profile, name, value, min_value, max_value, readonly); }
+    friend bool operator==(const SettingsProfileElement & lhs, const SettingsProfileElement & rhs) { return lhs.toTuple() == rhs.toTuple(); }
+    friend bool operator!=(const SettingsProfileElement & lhs, const SettingsProfileElement & rhs) { return !(lhs == rhs); }
+    friend bool operator <(const SettingsProfileElement & lhs, const SettingsProfileElement & rhs) { return lhs.toTuple() < rhs.toTuple(); }
+    friend bool operator >(const SettingsProfileElement & lhs, const SettingsProfileElement & rhs) { return rhs < lhs; }
+    friend bool operator <=(const SettingsProfileElement & lhs, const SettingsProfileElement & rhs) { return !(rhs < lhs); }
+    friend bool operator >=(const SettingsProfileElement & lhs, const SettingsProfileElement & rhs) { return !(lhs < rhs); }
+};
+
+
+class SettingsProfileElements : public std::vector<SettingsProfileElement>
+{
+public:
+    void merge(const SettingsProfileElements & other);
+
+    Settings toSettings() const;
+    SettingsChanges toSettingsChanges() const;
+    SettingsConstraints toSettingsConstraints() const;
+};
+
+}

dbms/src/Access/SettingsProfilesCache.cpp

+#include <Access/SettingsProfilesCache.h>
+#include <Access/AccessControlManager.h>
+#include <Access/SettingsProfile.h>
+#include <Common/quoteString.h>
+#include <boost/range/adaptor/map.hpp>
+#include <boost/range/algorithm_ext/push_back.hpp>
+
+
+namespace DB
+{
+namespace ErrorCodes
+{
+    extern const int THERE_IS_NO_PROFILE;
+}
+
+
+SettingsProfilesCache::SettingsProfilesCache(const AccessControlManager & manager_)
+    : manager(manager_) {}
+
+SettingsProfilesCache::~SettingsProfilesCache() = default;
+
+
+void SettingsProfilesCache::ensureAllProfilesRead()
+{
+    /// `mutex` is already locked.
+    if (all_profiles_read)
+        return;
+    all_profiles_read = true;
+
+    subscription = manager.subscribeForChanges<SettingsProfile>(
+        [&](const UUID & id, const AccessEntityPtr & entity)
+        {
+            if (entity)
+                profileAddedOrChanged(id, typeid_cast<SettingsProfilePtr>(entity));
+            else
+                profileRemoved(id);
+        });
+
+    for (const UUID & id : manager.findAll<SettingsProfile>())
+    {
+        auto profile = manager.tryRead<SettingsProfile>(id);
+        if (profile)
+        {
+            all_profiles.emplace(id, profile);
+            profiles_by_name[profile->getName()] = id;
+        }
+    }
+}
+
+
+void SettingsProfilesCache::profileAddedOrChanged(const UUID & profile_id, const SettingsProfilePtr & new_profile)
+{
+    std::lock_guard lock{mutex};
+    auto it = all_profiles.find(profile_id);
+    if (it == all_profiles.end())
+    {
+        all_profiles.emplace(profile_id, new_profile);
+        profiles_by_name[new_profile->getName()] = profile_id;
+    }
+    else
+    {
+        auto old_profile = it->second;
+        it->second = new_profile;
+        if (old_profile->getName() != new_profile->getName())
+            profiles_by_name.erase(old_profile->getName());
+        profiles_by_name[new_profile->getName()] = profile_id;
+    }
+    settings_for_profiles.clear();
+    mergeSettingsAndConstraints();
+}
+
+
+void SettingsProfilesCache::profileRemoved(const UUID & profile_id)
+{
+    std::lock_guard lock{mutex};
+    auto it = all_profiles.find(profile_id);
+    if (it == all_profiles.end())
+        return;
+    profiles_by_name.erase(it->second->getName());
+    all_profiles.erase(it);
+    settings_for_profiles.clear();
+    mergeSettingsAndConstraints();
+}
+
+
+void SettingsProfilesCache::setDefaultProfileName(const String & default_profile_name)
+{
+    std::lock_guard lock{mutex};
+    ensureAllProfilesRead();
+
+    if (default_profile_name.empty())
+    {
+        default_profile_id = {};
+        return;
+    }
+
+    auto it = profiles_by_name.find(default_profile_name);
+    if (it == profiles_by_name.end())
+        throw Exception("Settings profile " + backQuote(default_profile_name) + " not found", ErrorCodes::THERE_IS_NO_PROFILE);
+
+    default_profile_id = it->second;
+}
+
+void SettingsProfilesCache::mergeSettingsAndConstraints()
+{
+    /// `mutex` is already locked.
+    std::erase_if(
+        enabled_settings,
+        [&](const std::pair<EnabledSettings::Params, std::weak_ptr<EnabledSettings>> & pr)
+        {
+            auto enabled = pr.second.lock();
+            if (!enabled)
+                return true; // remove from the `enabled_settings` list.
+            mergeSettingsAndConstraintsFor(*enabled);
+            return false; // keep in the `enabled_settings` list.
+        });
+}
+
+
+void SettingsProfilesCache::mergeSettingsAndConstraintsFor(EnabledSettings & enabled) const
+{
+    SettingsProfileElements merged_settings;
+    if (default_profile_id)
+    {
+        SettingsProfileElement new_element;
+        new_element.parent_profile = *default_profile_id;
+        merged_settings.emplace_back(new_element);
+    }
+
+    for (const auto & [profile_id, profile] : all_profiles)
+        if (profile->to_roles.match(enabled.params.user_id, enabled.params.enabled_roles))
+        {
+            SettingsProfileElement new_element;
+            new_element.parent_profile = profile_id;
+            merged_settings.emplace_back(new_element);
+        }
+
+    merged_settings.merge(enabled.params.settings_from_enabled_roles);
+    merged_settings.merge(enabled.params.settings_from_user);
+
+    substituteProfiles(merged_settings);
+
+    enabled.setSettingsAndConstraints(
+                std::make_shared<Settings>(merged_settings.toSettings()),
+                std::make_shared<SettingsConstraints>(merged_settings.toSettingsConstraints()));
+}
+
+
+void SettingsProfilesCache::substituteProfiles(SettingsProfileElements & elements) const
+{
+    bool stop_substituting = false;
+    boost::container::flat_set<UUID> already_substituted;
+    while (!stop_substituting)
+    {
+        stop_substituting = true;
+        for (size_t i = 0; i != elements.size(); ++i)
+        {
+            auto & element = elements[i];
+            if (!element.parent_profile)
+                continue;
+
+            auto parent_profile_id = *element.parent_profile;
+            element.parent_profile.reset();
+            if (already_substituted.contains(parent_profile_id))
+                continue;
+
+            already_substituted.insert(parent_profile_id);
+            auto parent_profile = all_profiles.find(parent_profile_id);
+            if (parent_profile == all_profiles.end())
+                continue;
+
+            const auto & parent_profile_elements = parent_profile->second->elements;
+            elements.insert(elements.begin() + i + 1, parent_profile_elements.begin(), parent_profile_elements.end());
+            i += parent_profile_elements.size();
+            stop_substituting = false;
+        }
+    }
+}
+
+
+std::shared_ptr<const EnabledSettings> SettingsProfilesCache::getEnabledSettings(
+    const UUID & user_id,
+    const SettingsProfileElements & settings_from_user,
+    const std::vector<UUID> & enabled_roles,
+    const SettingsProfileElements & settings_from_enabled_roles)
+{
+    std::lock_guard lock{mutex};
+    ensureAllProfilesRead();
+
+    EnabledSettings::Params params;
+    params.user_id = user_id;
+    params.settings_from_user = settings_from_user;
+    params.enabled_roles = enabled_roles;
+    params.settings_from_enabled_roles = settings_from_enabled_roles;
+
+    auto it = enabled_settings.find(params);
+    if (it != enabled_settings.end())
+    {
+        auto from_cache = it->second.lock();
+        if (from_cache)
+            return from_cache;
+        enabled_settings.erase(it);
+    }
+
+    std::shared_ptr<EnabledSettings> res(new EnabledSettings(params));
+    enabled_settings.emplace(std::move(params), res);
+    mergeSettingsAndConstraintsFor(*res);
+    return res;
+}
+
+
+std::shared_ptr<const SettingsChanges> SettingsProfilesCache::getProfileSettings(const String & profile_name)
+{
+    std::lock_guard lock{mutex};
+    ensureAllProfilesRead();
+
+    auto it = profiles_by_name.find(profile_name);
+    if (it == profiles_by_name.end())
+        throw Exception("Settings profile " + backQuote(profile_name) + " not found", ErrorCodes::THERE_IS_NO_PROFILE);
+    const UUID profile_id = it->second;
+
+    auto it2 = settings_for_profiles.find(profile_id);
+    if (it2 != settings_for_profiles.end())
+        return it2->second;
+
+    SettingsProfileElements elements = all_profiles[profile_id]->elements;
+    substituteProfiles(elements);
+    auto res = std::make_shared<const SettingsChanges>(elements.toSettingsChanges());
+    settings_for_profiles.emplace(profile_id, res);
+    return res;
+}
+
+
+}

dbms/src/Access/SettingsProfilesCache.h

+#pragma once
+
+#include <Access/EnabledSettings.h>
+#include <Core/UUID.h>
+#include <Core/Types.h>
+#include <ext/scope_guard.h>
+#include <map>
+#include <unordered_map>
+
+
+namespace DB
+{
+class AccessControlManager;
+struct SettingsProfile;
+using SettingsProfilePtr = std::shared_ptr<const SettingsProfile>;
+class SettingsProfileElements;
+class EnabledSettings;
+
+
+/// Reads and caches all the settings profiles.
+class SettingsProfilesCache
+{
+public:
+    SettingsProfilesCache(const AccessControlManager & manager_);
+    ~SettingsProfilesCache();
+
+    void setDefaultProfileName(const String & default_profile_name);
+
+    std::shared_ptr<const EnabledSettings> getEnabledSettings(
+        const UUID & user_id,
+        const SettingsProfileElements & settings_from_user_,
+        const std::vector<UUID> & enabled_roles,
+        const SettingsProfileElements & settings_from_enabled_roles_);
+
+    std::shared_ptr<const SettingsChanges> getProfileSettings(const String & profile_name);
+
+private:
+    void ensureAllProfilesRead();
+    void profileAddedOrChanged(const UUID & profile_id, const SettingsProfilePtr & new_profile);
+    void profileRemoved(const UUID & profile_id);
+    void mergeSettingsAndConstraints();
+    void mergeSettingsAndConstraintsFor(EnabledSettings & enabled) const;
+    void substituteProfiles(SettingsProfileElements & elements) const;
+
+    const AccessControlManager & manager;
+    std::unordered_map<UUID, SettingsProfilePtr> all_profiles;
+    std::unordered_map<String, UUID> profiles_by_name;
+    bool all_profiles_read = false;
+    ext::scope_guard subscription;
+    std::map<EnabledSettings::Params, std::weak_ptr<EnabledSettings>> enabled_settings;
+    std::optional<UUID> default_profile_id;
+    std::unordered_map<UUID, std::shared_ptr<const SettingsChanges>> settings_for_profiles;
+    mutable std::mutex mutex;
+};
+}

dbms/src/Access/User.cpp

@@ -12,7 +12,7 @@ bool User::equal(const IAccessEntity & other) const
     return (authentication == other_user.authentication) && (allowed_client_hosts == other_user.allowed_client_hosts)
         && (access == other_user.access) && (access_with_grant_option == other_user.access_with_grant_option)
         && (granted_roles == other_user.granted_roles) && (granted_roles_with_admin_option == other_user.granted_roles_with_admin_option)
-        && (default_roles == other_user.default_roles) && (profile == other_user.profile);
+        && (default_roles == other_user.default_roles) && (settings == other_user.settings);
 }
 
 }

dbms/src/Access/User.h

@@ -5,6 +5,7 @@
 #include <Access/AllowedClientHosts.h>
 #include <Access/AccessRights.h>
 #include <Access/ExtendedRoleSet.h>
+#include <Access/SettingsProfileElement.h>
 #include <Core/UUID.h>
 #include <boost/container/flat_set.hpp>
 
@@ -22,7 +23,7 @@ struct User : public IAccessEntity
     boost::container::flat_set<UUID> granted_roles;
     boost::container::flat_set<UUID> granted_roles_with_admin_option;
     ExtendedRoleSet default_roles = ExtendedRoleSet::AllTag{};
-    String profile;
+    SettingsProfileElements settings;
 
     bool equal(const IAccessEntity & other) const override;
     std::shared_ptr<IAccessEntity> clone() const override { return cloneImpl<User>(); }

dbms/src/Access/UsersConfigAccessStorage.cpp

@@ -2,11 +2,15 @@
 #include <Access/Quota.h>
 #include <Access/RowPolicy.h>
 #include <Access/User.h>
+#include <Access/SettingsProfile.h>
 #include <Dictionaries/IDictionary.h>
 #include <Common/StringUtils/StringUtils.h>
 #include <Common/quoteString.h>
 #include <Poco/Util/AbstractConfiguration.h>
 #include <Poco/MD5Engine.h>
+#include <common/logger_useful.h>
+#include <boost/range/algorithm/copy.hpp>
+#include <boost/range/adaptor/map.hpp>
 #include <cstring>
 
 
@@ -16,6 +20,7 @@ namespace ErrorCodes
 {
     extern const int BAD_ARGUMENTS;
     extern const int UNKNOWN_ADDRESS_PATTERN_TYPE;
+    extern const int NOT_IMPLEMENTED;
 }
 
 
@@ -29,6 +34,8 @@ namespace
             return 'Q';
         if (type == typeid(RowPolicy))
             return 'P';
+        if (type == typeid(SettingsProfile))
+            return 'S';
         return 0;
     }
 
@@ -82,7 +89,14 @@ namespace
             user->authentication.setPasswordHashHex(config.getString(user_config + ".password_double_sha1_hex"));
         }
 
-        user->profile = config.getString(user_config + ".profile");
+        const auto profile_name_config = user_config + ".profile";
+        if (config.has(profile_name_config))
+        {
+            auto profile_name = config.getString(profile_name_config);
+            SettingsProfileElement profile_element;
+            profile_element.parent_profile = generateID(typeid(SettingsProfile), profile_name);
+            user->settings.push_back(std::move(profile_element));
+        }
 
         /// Fill list of allowed hosts.
         const auto networks_config = user_config + ".networks";
@@ -330,6 +344,93 @@ namespace
         }
         return policies;
     }
+
+
+    SettingsProfileElements parseSettingsConstraints(const Poco::Util::AbstractConfiguration & config,
+                                                     const String & path_to_constraints)
+    {
+        SettingsProfileElements profile_elements;
+        Poco::Util::AbstractConfiguration::Keys names;
+        config.keys(path_to_constraints, names);
+        for (const String & name : names)
+        {
+            SettingsProfileElement profile_element;
+            profile_element.name = name;
+            Poco::Util::AbstractConfiguration::Keys constraint_types;
+            String path_to_name = path_to_constraints + "." + name;
+            config.keys(path_to_name, constraint_types);
+            for (const String & constraint_type : constraint_types)
+            {
+                if (constraint_type == "min")
+                    profile_element.min_value = config.getString(path_to_name + "." + constraint_type);
+                else if (constraint_type == "max")
+                    profile_element.max_value = config.getString(path_to_name + "." + constraint_type);
+                else if (constraint_type == "readonly")
+                    profile_element.readonly = true;
+                else
+                    throw Exception("Setting " + constraint_type + " value for " + name + " isn't supported", ErrorCodes::NOT_IMPLEMENTED);
+            }
+            profile_elements.push_back(std::move(profile_element));
+        }
+        return profile_elements;
+    }
+
+    std::shared_ptr<SettingsProfile> parseSettingsProfile(
+        const Poco::Util::AbstractConfiguration & config,
+        const String & profile_name)
+    {
+        auto profile = std::make_shared<SettingsProfile>();
+        profile->setName(profile_name);
+        String profile_config = "profiles." + profile_name;
+
+        Poco::Util::AbstractConfiguration::Keys keys;
+        config.keys(profile_config, keys);
+
+        for (const std::string & key : keys)
+        {
+            if (key == "profile" || key.starts_with("profile["))
+            {
+                String parent_profile_name = config.getString(profile_config + "." + key);
+                SettingsProfileElement profile_element;
+                profile_element.parent_profile = generateID(typeid(SettingsProfile), parent_profile_name);
+                profile->elements.emplace_back(std::move(profile_element));
+                continue;
+            }
+
+            if (key == "constraints" || key.starts_with("constraints["))
+            {
+                profile->elements.merge(parseSettingsConstraints(config, profile_config + "." + key));
+                continue;
+            }
+
+            SettingsProfileElement profile_element;
+            profile_element.name = key;
+            profile_element.value = config.getString(profile_config + "." + key);
+            profile->elements.emplace_back(std::move(profile_element));
+        }
+
+        return profile;
+    }
+
+
+    std::vector<AccessEntityPtr> parseSettingsProfiles(const Poco::Util::AbstractConfiguration & config, Poco::Logger * log)
+    {
+        std::vector<AccessEntityPtr> profiles;
+        Poco::Util::AbstractConfiguration::Keys profile_names;
+        config.keys("profiles", profile_names);
+        for (const auto & profile_name : profile_names)
+        {
+            try
+            {
+                profiles.push_back(parseSettingsProfile(config, profile_name));
+            }
+            catch (...)
+            {
+                tryLogCurrentException(log, "Could not parse profile " + backQuote(profile_name));
+            }
+        }
+        return profiles;
+    }
 }
 
 
@@ -347,6 +448,8 @@ void UsersConfigAccessStorage::setConfiguration(const Poco::Util::AbstractConfig
         all_entities.emplace_back(generateID(*entity), entity);
     for (const auto & entity : parseRowPolicies(config, getLogger()))
         all_entities.emplace_back(generateID(*entity), entity);
+    for (const auto & entity : parseSettingsProfiles(config, getLogger()))
+        all_entities.emplace_back(generateID(*entity), entity);
     memory_storage.setAll(all_entities);
 }
 

dbms/src/Interpreters/Context.cpp

@@ -30,6 +30,7 @@
 #include <Access/ContextAccess.h>
 #include <Access/EnabledRowPolicies.h>
 #include <Access/User.h>
+#include <Access/SettingsProfile.h>
 #include <Access/SettingsConstraints.h>
 #include <Interpreters/ExpressionJIT.h>
 #include <Dictionaries/Embedded/GeoDictionariesLoader.h>
@@ -633,7 +634,7 @@ void Context::setUser(const String & name, const String & password, const Poco::
     std::shared_ptr<const ContextAccess> new_access;
     if (new_user_id)
     {
-        new_access = getAccessControlManager().getContextAccess(*new_user_id, {}, true, settings, current_database, client_info);
+        new_access = getAccessControlManager().getContextAccess(*new_user_id, {}, true, {}, current_database, client_info);
         if (!new_access->isClientHostAllowed() || !new_access->isCorrectPassword(password))
         {
             new_user_id = {};
@@ -649,7 +650,7 @@ void Context::setUser(const String & name, const String & password, const Poco::
     current_roles.clear();
     use_default_roles = true;
 
-    calculateUserSettings();
+    setSettings(*access->getDefaultSettings());
 }
 
 std::shared_ptr<const User> Context::getUser() const
@@ -776,42 +777,9 @@ std::shared_ptr<const EnabledQuota> Context::getQuota() const
 }
 
 
-void Context::calculateUserSettings()
+void Context::setProfile(const String & profile_name)
 {
-    auto lock = getLock();
-    String profile = getUser()->profile;
-
-    bool old_readonly = settings.readonly;
-    bool old_allow_ddl = settings.allow_ddl;
-    bool old_allow_introspection_functions = settings.allow_introspection_functions;
-
-    /// 1) Set default settings (hardcoded values)
-    /// NOTE: we ignore global_context settings (from which it is usually copied)
-    /// NOTE: global_context settings are immutable and not auto updated
-    settings = Settings();
-    settings_constraints = nullptr;
-
-    /// 2) Apply settings from default profile
-    auto default_profile_name = getDefaultProfileName();
-    if (profile != default_profile_name)
-        setProfile(default_profile_name);
-
-    /// 3) Apply settings from current user
-    setProfile(profile);
-
-    /// 4) Recalculate access rights if it's necessary.
-    if ((settings.readonly != old_readonly) || (settings.allow_ddl != old_allow_ddl) || (settings.allow_introspection_functions != old_allow_introspection_functions))
-        calculateAccessRights();
-}
-
-void Context::setProfile(const String & profile)
-{
-    settings.setProfile(profile, *shared->users_config);
-
-    auto new_constraints
-        = settings_constraints ? std::make_shared<SettingsConstraints>(*settings_constraints) : std::make_shared<SettingsConstraints>();
-    new_constraints->setProfile(profile, *shared->users_config);
-    settings_constraints = std::move(new_constraints);
+    applySettingsChanges(*getAccessControlManager().getProfileSettings(profile_name));
 }
 
 
@@ -993,30 +961,37 @@ void Context::applySettingsChanges(const SettingsChanges & changes)
 
 void Context::checkSettingsConstraints(const SettingChange & change) const
 {
-    if (settings_constraints)
+    if (auto settings_constraints = getSettingsConstraints())
         settings_constraints->check(settings, change);
 }
 
 void Context::checkSettingsConstraints(const SettingsChanges & changes) const
 {
-    if (settings_constraints)
+    if (auto settings_constraints = getSettingsConstraints())
         settings_constraints->check(settings, changes);
 }
 
 
 void Context::clampToSettingsConstraints(SettingChange & change) const
 {
-    if (settings_constraints)
+    if (auto settings_constraints = getSettingsConstraints())
         settings_constraints->clamp(settings, change);
 }
 
 void Context::clampToSettingsConstraints(SettingsChanges & changes) const
 {
-    if (settings_constraints)
+    if (auto settings_constraints = getSettingsConstraints())
         settings_constraints->clamp(settings, changes);
 }
 
 
+std::shared_ptr<const SettingsConstraints> Context::getSettingsConstraints() const
+{
+    auto lock = getLock();
+    return access->getSettingsConstraints();
+}
+
+
 String Context::getCurrentDatabase() const
 {
     auto lock = getLock();
@@ -1877,8 +1852,10 @@ void Context::setApplicationType(ApplicationType type)
 void Context::setDefaultProfiles(const Poco::Util::AbstractConfiguration & config)
 {
     shared->default_profile_name = config.getString("default_profile", "default");
+    getAccessControlManager().setDefaultProfileName(shared->default_profile_name);
+
     shared->system_profile_name = config.getString("system_profile", shared->default_profile_name);
-    setSetting("profile", shared->system_profile_name);
+    setProfile(shared->system_profile_name);
 }
 
 String Context::getDefaultProfileName() const

dbms/src/Interpreters/Context.h

@@ -153,7 +153,6 @@ private:
     std::shared_ptr<const EnabledRowPolicies> initial_row_policy;
     String current_database;
     Settings settings;                                  /// Setting for query execution.
-    std::shared_ptr<const SettingsConstraints> settings_constraints;
     using ProgressCallback = std::function<void(const Progress & progress)>;
     ProgressCallback progress_callback;                 /// Callback for tracking progress of query execution.
     QueryStatus * process_list_elem = nullptr;   /// For tracking total resource usage for query.
@@ -353,7 +352,7 @@ public:
     void clampToSettingsConstraints(SettingsChanges & changes) const;
 
     /// Returns the current constraints (can return null).
-    std::shared_ptr<const SettingsConstraints> getSettingsConstraints() const { return settings_constraints; }
+    std::shared_ptr<const SettingsConstraints> getSettingsConstraints() const;
 
     const EmbeddedDictionaries & getEmbeddedDictionaries() const;
     const ExternalDictionariesLoader & getExternalDictionariesLoader() const;
@@ -593,7 +592,6 @@ private:
     std::unique_lock<std::recursive_mutex> getLock() const;
 
     /// Compute and set actual user settings, client_info.current_user should be set
-    void calculateUserSettings();
     void calculateAccessRights();
 
     template <typename... Args>

dbms/src/Interpreters/InterpreterCreateUserQuery.cpp

@@ -47,9 +47,6 @@ namespace
 
             InterpreterSetRoleQuery::updateUserSetDefaultRoles(user, *default_roles);
         }
-
-        if (query.profile)
-            user.profile = *query.profile;
     }
 }
 

dbms/src/Interpreters/InterpreterShowCreateAccessEntityQuery.cpp

@@ -43,9 +43,6 @@ namespace
         if (user.allowed_client_hosts != AllowedClientHosts::AnyHostTag{})
             query->hosts = user.allowed_client_hosts;
 
-        if (!user.profile.empty())
-            query->profile = user.profile;
-
         if (user.default_roles != ExtendedRoleSet::AllTag{})
         {
             if (attach_mode)