Merge 396544a70d into 44b4bd38b9

Enable cluster table functions for DataLake Storages

docs/en/sql-reference/statements/alter/role.md

@@ -13,5 +13,10 @@ Syntax:
 ``` sql
 ALTER ROLE [IF EXISTS] name1 [RENAME TO new_name |, name2 [,...]] 
     [ON CLUSTER cluster_name]
-    [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+    [ADD|MODIFY SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+    [DROP SETTINGS variable [,...] ]
+    [ADD PROFILES 'profile_name' [,...] ]
+    [DROP PROFILES 'profile_name' [,...] ]
+    [DROP ALL PROFILES]
+    [DROP ALL SETTINGS]
 ```

docs/en/sql-reference/statements/alter/settings-profile.md

@@ -13,6 +13,11 @@ Syntax:
 ``` sql
 ALTER SETTINGS PROFILE [IF EXISTS] name1 [RENAME TO new_name |, name2 [,...]] 
     [ON CLUSTER cluster_name]
-    [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | INHERIT 'profile_name'] [,...]
+    [ADD|MODIFY SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | INHERIT 'profile_name'] [,...]
     [TO {{role1 | user1 [, role2 | user2 ...]} | NONE | ALL | ALL EXCEPT {role1 | user1 [, role2 | user2 ...]}}]
+    [DROP SETTINGS variable [,...] ]
+    [ADD PROFILES 'profile_name' [,...] ]
+    [DROP PROFILES 'profile_name' [,...] ]
+    [DROP ALL PROFILES]
+    [DROP ALL SETTINGS]
 ```

docs/en/sql-reference/statements/alter/user.md

@@ -18,7 +18,12 @@ ALTER USER [IF EXISTS] name1 [RENAME TO new_name |, name2 [,...]]
     [VALID UNTIL datetime]
     [DEFAULT ROLE role [,...] | ALL | ALL EXCEPT role [,...] ]
     [GRANTEES {user | role | ANY | NONE} [,...] [EXCEPT {user | role} [,...]]]
-    [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [READONLY | WRITABLE] | PROFILE 'profile_name'] [,...]
+    [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [READONLY|WRITABLE|CONST|CHANGEABLE_IN_READONLY] [,...] ]
+    [DROP SETTINGS variable [,...] ]
+    [ADD PROFILES 'profile_name' [,...] ]
+    [DROP PROFILES 'profile_name' [,...] ]
+    [DROP ALL PROFILES]
+    [DROP ALL SETTINGS]
 ```
 
 To use `ALTER USER` you must have the [ALTER USER](../../../sql-reference/statements/grant.md#access-management) privilege.

docs/en/sql-reference/table-functions/deltalake.md

@@ -49,4 +49,4 @@ LIMIT 2
 **See Also**
 
 - [DeltaLake engine](/docs/en/engines/table-engines/integrations/deltalake.md)
-
+- [DeltaLake cluster table function](/docs/en/sql-reference/table-functions/deltalakeCluster.md)

docs/en/sql-reference/table-functions/deltalakeCluster.md

@@ -0,0 +1,30 @@
+---
+slug: /en/sql-reference/table-functions/deltalakeCluster
+sidebar_position: 46
+sidebar_label: deltaLakeCluster
+title: "deltaLakeCluster Table Function"
+---
+This is an extension to the [deltaLake](/docs/en/sql-reference/table-functions/deltalake.md) table function.
+
+Allows processing files from [Delta Lake](https://github.com/delta-io/delta) tables in Amazon S3 in parallel from many nodes in a specified cluster. On initiator it creates a connection to all nodes in the cluster and dispatches each file dynamically. On the worker node it asks the initiator about the next task to process and processes it. This is repeated until all tasks are finished.
+
+**Syntax**
+
+``` sql
+deltaLakeCluster(cluster_name, url [,aws_access_key_id, aws_secret_access_key] [,format] [,structure] [,compression])
+```
+
+**Arguments**
+
+- `cluster_name` — Name of a cluster that is used to build a set of addresses and connection parameters to remote and local servers.
+
+- Description of all other arguments coincides with description of arguments in equivalent [deltaLake](/docs/en/sql-reference/table-functions/deltalake.md) table function.
+
+**Returned value**
+
+A table with the specified structure for reading data from cluster in the specified Delta Lake table in S3.
+
+**See Also**
+
+- [deltaLake engine](/docs/en/engines/table-engines/integrations/deltalake.md)
+- [deltaLake table function](/docs/en/sql-reference/table-functions/deltalake.md)

docs/en/sql-reference/table-functions/hudi.md

@@ -29,4 +29,4 @@ A table with the specified structure for reading data in the specified Hudi tabl
 **See Also**
 
 - [Hudi engine](/docs/en/engines/table-engines/integrations/hudi.md)
-
+- [Hudi cluster table function](/docs/en/sql-reference/table-functions/hudiCluster.md)

docs/en/sql-reference/table-functions/hudiCluster.md

@@ -0,0 +1,30 @@
+---
+slug: /en/sql-reference/table-functions/hudiCluster
+sidebar_position: 86
+sidebar_label: hudiCluster
+title: "hudiCluster Table Function"
+---
+This is an extension to the [hudi](/docs/en/sql-reference/table-functions/hudi.md) table function.
+
+Allows processing files from Apache [Hudi](https://hudi.apache.org/) tables in Amazon S3 in parallel from many nodes in a specified cluster. On initiator it creates a connection to all nodes in the cluster and dispatches each file dynamically. On the worker node it asks the initiator about the next task to process and processes it. This is repeated until all tasks are finished.
+
+**Syntax**
+
+``` sql
+hudiCluster(cluster_name, url [,aws_access_key_id, aws_secret_access_key] [,format] [,structure] [,compression])
+```
+
+**Arguments**
+
+- `cluster_name` — Name of a cluster that is used to build a set of addresses and connection parameters to remote and local servers.
+
+- Description of all other arguments coincides with description of arguments in equivalent [hudi](/docs/en/sql-reference/table-functions/hudi.md) table function.
+
+**Returned value**
+
+A table with the specified structure for reading data from cluster in the specified Hudi table in S3.
+
+**See Also**
+
+- [Hudi engine](/docs/en/engines/table-engines/integrations/hudi.md)
+- [Hudi table function](/docs/en/sql-reference/table-functions/hudi.md)

docs/en/sql-reference/table-functions/iceberg.md

@@ -72,3 +72,4 @@ Table function `iceberg` is an alias to `icebergS3` now.
 **See Also**
 
 - [Iceberg engine](/docs/en/engines/table-engines/integrations/iceberg.md)
+- [Iceberg cluster table function](/docs/en/sql-reference/table-functions/icebergCluster.md)

docs/en/sql-reference/table-functions/icebergCluster.md

@@ -0,0 +1,43 @@
+---
+slug: /en/sql-reference/table-functions/icebergCluster
+sidebar_position: 91
+sidebar_label: icebergCluster
+title: "icebergCluster Table Function"
+---
+This is an extension to the [iceberg](/docs/en/sql-reference/table-functions/iceberg.md) table function.
+
+Allows processing files from Apache [Iceberg](https://iceberg.apache.org/) in parallel from many nodes in a specified cluster. On initiator it creates a connection to all nodes in the cluster and dispatches each file dynamically. On the worker node it asks the initiator about the next task to process and processes it. This is repeated until all tasks are finished.
+
+**Syntax**
+
+``` sql
+icebergS3Cluster(cluster_name, url [, NOSIGN | access_key_id, secret_access_key, [session_token]] [,format] [,compression_method])
+icebergS3Cluster(cluster_name, named_collection[, option=value [,..]])
+
+icebergAzureCluster(cluster_name, connection_string|storage_account_url, container_name, blobpath, [,account_name], [,account_key] [,format] [,compression_method])
+icebergAzureCluster(cluster_name, named_collection[, option=value [,..]])
+
+icebergHDFSCluster(cluster_name, path_to_table, [,format] [,compression_method])
+icebergHDFSCluster(cluster_name, named_collection[, option=value [,..]])
+```
+
+**Arguments**
+
+- `cluster_name` — Name of a cluster that is used to build a set of addresses and connection parameters to remote and local servers.
+
+- Description of all other arguments coincides with description of arguments in equivalent [iceberg](/docs/en/sql-reference/table-functions/iceberg.md) table function.
+
+**Returned value**
+
+A table with the specified structure for reading data from cluster in the specified Iceberg table.
+
+**Examples**
+
+```sql
+SELECT * FROM icebergS3Cluster('cluster_simple', 'http://test.s3.amazonaws.com/clickhouse-bucket/test_table', 'test', 'test')
+```
+
+**See Also**
+
+- [Iceberg engine](/docs/en/engines/table-engines/integrations/iceberg.md)
+- [Iceberg table function](/docs/en/sql-reference/table-functions/iceberg.md)

docs/ru/sql-reference/statements/alter/role.md

@@ -13,5 +13,10 @@ sidebar_label: ROLE
 ``` sql
 ALTER ROLE [IF EXISTS] name1 [RENAME TO new_name |, name2 [,...]] 
     [ON CLUSTER cluster_name]
-    [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+    [ADD|MODIFY SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+    [DROP SETTINGS variable [,...] ]
+    [ADD PROFILES 'profile_name' [,...] ]
+    [DROP PROFILES 'profile_name' [,...] ]
+    [DROP ALL PROFILES]
+    [DROP ALL SETTINGS]
 ```

docs/ru/sql-reference/statements/alter/settings-profile.md

@@ -13,6 +13,11 @@ sidebar_label: SETTINGS PROFILE
 ``` sql
 ALTER SETTINGS PROFILE [IF EXISTS] name1 [RENAME TO new_name |, name2 [,...]] 
     [ON CLUSTER cluster_name]
-    [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | INHERIT 'profile_name'] [,...]
+    [ADD|MODIFY SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | INHERIT 'profile_name'] [,...]
     [TO {{role1 | user1 [, role2 | user2 ...]} | NONE | ALL | ALL EXCEPT {role1 | user1 [, role2 | user2 ...]}}]
+    [DROP SETTINGS variable [,...] ]
+    [ADD PROFILES 'profile_name' [,...] ]
+    [DROP PROFILES 'profile_name' [,...] ]
+    [DROP ALL PROFILES]
+    [DROP ALL SETTINGS]
 ```

docs/ru/sql-reference/statements/alter/user.md

@@ -19,7 +19,12 @@ ALTER USER [IF EXISTS] name1 [RENAME TO new_name |, name2 [,...]]
     [VALID UNTIL datetime]
     [DEFAULT ROLE role [,...] | ALL | ALL EXCEPT role [,...] ]
     [GRANTEES {user | role | ANY | NONE} [,...] [EXCEPT {user | role} [,...]]]
-    [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [READONLY | WRITABLE] | PROFILE 'profile_name'] [,...]
+    [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [READONLY|WRITABLE|CONST|CHANGEABLE_IN_READONLY] [,...] ]
+    [DROP SETTINGS variable [,...] ]
+    [ADD PROFILES 'profile_name' [,...] ]
+    [DROP PROFILES 'profile_name' [,...] ]
+    [DROP ALL PROFILES]
+    [DROP ALL SETTINGS]
 ```
 
 Для выполнения `ALTER USER` необходима привилегия [ALTER USER](../grant.md#grant-access-management).

src/Access/RoleCache.cpp

@@ -45,7 +45,7 @@ namespace
 
         roles_info.names_of_roles[role_id] = role->getName();
         roles_info.access.makeUnion(role->access);
-        roles_info.settings_from_enabled_roles.merge(role->settings);
+        roles_info.settings_from_enabled_roles.merge(role->settings, /* normalize= */ false);
 
         for (const auto & granted_role : role->granted_roles.getGranted())
             collectRoles(roles_info, skip_ids, get_role_function, granted_role, false, false);

src/Access/SettingsConstraints.cpp

@@ -137,6 +137,13 @@ void SettingsConstraints::merge(const SettingsConstraints & other)
 }
 
 
+void SettingsConstraints::check(const Settings & current_settings, const AlterSettingsProfileElements & profile_elements, SettingSource source) const
+{
+    check(current_settings, profile_elements.add_settings, source);
+    check(current_settings, profile_elements.modify_settings, source);
+    /// We don't check `drop_settings` here.
+}
+
 void SettingsConstraints::check(const Settings & current_settings, const SettingsProfileElements & profile_elements, SettingSource source) const
 {
     for (const auto & element : profile_elements)

src/Access/SettingsConstraints.h

@@ -74,10 +74,11 @@ public:
     void merge(const SettingsConstraints & other);
 
     /// Checks whether `change` violates these constraints and throws an exception if so.
-    void check(const Settings & current_settings, const SettingsProfileElements & profile_elements, SettingSource source) const;
     void check(const Settings & current_settings, const SettingChange & change, SettingSource source) const;
     void check(const Settings & current_settings, const SettingsChanges & changes, SettingSource source) const;
     void check(const Settings & current_settings, SettingsChanges & changes, SettingSource source) const;
+    void check(const Settings & current_settings, const SettingsProfileElements & profile_elements, SettingSource source) const;
+    void check(const Settings & current_settings, const AlterSettingsProfileElements & profile_elements, SettingSource source) const;
 
     /// Checks whether `change` violates these constraints and throws an exception if so. (setting short name is expected inside `changes`)
     void check(const MergeTreeSettings & current_settings, const SettingChange & change) const;

src/Access/SettingsProfileElement.cpp

@@ -9,6 +9,8 @@
 #include <IO/WriteHelpers.h>
 #include <Parsers/Access/ASTSettingsProfileElement.h>
 #include <base/removeDuplicates.h>
+#include <boost/container/flat_map.hpp>
+#include <boost/container/flat_set.hpp>
 
 
 namespace DB
@@ -19,6 +21,7 @@ namespace ErrorCodes
     extern const int NOT_IMPLEMENTED;
 }
 
+
 SettingsProfileElement::SettingsProfileElement(const ASTSettingsProfileElement & ast)
 {
     init(ast, nullptr);
@@ -116,16 +119,20 @@ std::shared_ptr<ASTSettingsProfileElement> SettingsProfileElement::toASTWithName
 }
 
 
-SettingsProfileElements::SettingsProfileElements(const ASTSettingsProfileElements & ast)
+SettingsProfileElements::SettingsProfileElements(const ASTSettingsProfileElements & ast, bool normalize_)
 {
     for (const auto & ast_element : ast.elements)
         emplace_back(*ast_element);
+    if (normalize_)
+        normalize();
 }
 
-SettingsProfileElements::SettingsProfileElements(const ASTSettingsProfileElements & ast, const AccessControl & access_control)
+SettingsProfileElements::SettingsProfileElements(const ASTSettingsProfileElements & ast, const AccessControl & access_control, bool normalize_)
 {
     for (const auto & ast_element : ast.elements)
         emplace_back(*ast_element, access_control);
+    if (normalize_)
+        normalize();
 }
 
 
@@ -133,7 +140,11 @@ std::shared_ptr<ASTSettingsProfileElements> SettingsProfileElements::toAST() con
 {
     auto res = std::make_shared<ASTSettingsProfileElements>();
     for (const auto & element : *this)
-        res->elements.push_back(element.toAST());
+    {
+        auto element_ast = element.toAST();
+        if (!element_ast->empty())
+            res->elements.push_back(element_ast);
+    }
     return res;
 }
 
@@ -141,7 +152,11 @@ std::shared_ptr<ASTSettingsProfileElements> SettingsProfileElements::toASTWithNa
 {
     auto res = std::make_shared<ASTSettingsProfileElements>();
     for (const auto & element : *this)
-        res->elements.push_back(element.toASTWithNames(access_control));
+    {
+        auto element_ast = element.toASTWithNames(access_control);
+        if (!element_ast->empty())
+            res->elements.push_back(element_ast);
+    }
     return res;
 }
 
@@ -220,9 +235,11 @@ void SettingsProfileElements::removeSettingsKeepProfiles()
 }
 
 
-void SettingsProfileElements::merge(const SettingsProfileElements & other)
+void SettingsProfileElements::merge(const SettingsProfileElements & other, bool normalize_)
 {
     insert(end(), other.begin(), other.end());
+    if (normalize_)
+        normalize();
 }
 
 
@@ -280,6 +297,81 @@ std::vector<UUID> SettingsProfileElements::toProfileIDs() const
     return res;
 }
 
+
+void SettingsProfileElements::normalize()
+{
+    /// Ensure that each element represents either a setting or a profile.
+    {
+        SettingsProfileElements new_elements;
+        for (auto & element : *this)
+        {
+            if (element.parent_profile && !element.setting_name.empty())
+            {
+                SettingsProfileElement new_element;
+                new_element.parent_profile = element.parent_profile;
+                element.parent_profile.reset();
+                new_elements.push_back(std::move(new_element));
+            }
+        }
+        insert(end(), new_elements.begin(), new_elements.end());
+    }
+
+    /// Partitioning: first profiles, then settings.
+    /// We use std::stable_partition() here because we want to preserve the relative order of profiles and the relative order of settings.
+    /// (We need that order to be preserved to remove duplicates correctly - see below.)
+    auto profiles_begin = begin();
+    auto profiles_end = std::stable_partition(begin(), end(), [](const SettingsProfileElement & element) { return static_cast<bool>(element.parent_profile); });
+    auto settings_begin = profiles_end;
+    auto settings_end = end();
+
+    /// Remove duplicates among profiles.
+    /// We keep the last position of any used profile.
+    /// It's important to keep exactly the last position (and not just any position) because profiles can override settings from each other.
+    /// For example, [pr_A, pr_B, pr_A, pr_C] is always the same as [pr_B, pr_A, pr_C], but can be not the same as [pr_A, pr_B, pr_C]
+    /// if pr_A and pr_B give different values to same settings.
+    {
+        boost::container::flat_set<UUID> profile_ids;
+        profile_ids.reserve(profiles_end - profiles_begin);
+        auto it = profiles_end;
+        while (it != profiles_begin)
+        {
+            --it;
+            auto & element = *it;
+            if (element.parent_profile && !profile_ids.emplace(*element.parent_profile).second)
+                element.parent_profile.reset();
+        }
+    }
+
+    /// Remove duplicates among settings.
+    /// We keep the first position of any used setting, and merge settings with the same name to that first element.
+    {
+        boost::container::flat_map<std::string_view, SettingsProfileElements::iterator> setting_name_to_first_encounter;
+        setting_name_to_first_encounter.reserve(settings_end - settings_begin);
+        for (auto it = settings_begin; it != settings_end; ++it)
+        {
+            auto & element = *it;
+            auto first = setting_name_to_first_encounter.emplace(element.setting_name, it).first->second;
+            if (it != first)
+            {
+                auto & first_element = *first;
+                if (element.value)
+                    first_element.value = element.value;
+                if (element.min_value)
+                    first_element.min_value = element.min_value;
+                if (element.max_value)
+                    first_element.max_value = element.max_value;
+                if (element.writability)
+                    first_element.writability = element.writability;
+                element.setting_name.clear();
+            }
+        }
+    }
+
+    /// Remove empty elements.
+    std::erase_if(*this, [](const SettingsProfileElement & element) { return element.empty(); });
+}
+
+
 bool SettingsProfileElements::isBackupAllowed() const
 {
     for (const auto & setting : *this)
@@ -296,4 +388,139 @@ bool SettingsProfileElements::isAllowBackupSetting(const String & setting_name)
     return Settings::resolveName(setting_name) == ALLOW_BACKUP_SETTING_NAME;
 }
 
+
+AlterSettingsProfileElements::AlterSettingsProfileElements(const SettingsProfileElements & ast)
+{
+    drop_all_settings = true;
+    drop_all_profiles = true;
+    add_settings = ast;
+}
+
+AlterSettingsProfileElements::AlterSettingsProfileElements(const ASTSettingsProfileElements & ast)
+    : AlterSettingsProfileElements(SettingsProfileElements{ast})
+{
+}
+
+AlterSettingsProfileElements::AlterSettingsProfileElements(const ASTSettingsProfileElements & ast, const AccessControl & access_control)
+    : AlterSettingsProfileElements(SettingsProfileElements{ast, access_control})
+{
+}
+
+AlterSettingsProfileElements::AlterSettingsProfileElements(const ASTAlterSettingsProfileElements & ast)
+{
+    drop_all_settings = ast.drop_all_settings;
+    drop_all_profiles = ast.drop_all_profiles;
+
+    if (ast.add_settings)
+        add_settings = SettingsProfileElements{*ast.add_settings, /* normalize= */ false}; /// For "ALTER" the normalization is unnecessary.
+
+    if (ast.modify_settings)
+        modify_settings = SettingsProfileElements{*ast.modify_settings, /* normalize= */ false};
+
+    if (ast.drop_settings)
+        drop_settings = SettingsProfileElements{*ast.drop_settings, /* normalize= */ false};
+}
+
+AlterSettingsProfileElements::AlterSettingsProfileElements(const ASTAlterSettingsProfileElements & ast, const AccessControl & access_control)
+{
+    drop_all_settings = ast.drop_all_settings;
+    drop_all_profiles = ast.drop_all_profiles;
+
+    if (ast.add_settings)
+        add_settings = SettingsProfileElements{*ast.add_settings, access_control, /* normalize= */ false}; /// For "ALTER" the normalization is unnecessary.
+
+    if (ast.modify_settings)
+        modify_settings = SettingsProfileElements{*ast.modify_settings, access_control, /* normalize= */ false};
+
+    if (ast.drop_settings)
+        drop_settings = SettingsProfileElements{*ast.drop_settings, access_control, /* normalize= */ false};
+}
+
+void SettingsProfileElements::applyChanges(const AlterSettingsProfileElements & changes)
+{
+    /// Apply "DROP" changes.
+    if (changes.drop_all_profiles)
+    {
+        for (auto & element : *this)
+            element.parent_profile.reset(); /// We only make this element empty, the element will be removed in normalizeProfileElements().
+    }
+
+    if (changes.drop_all_settings)
+    {
+        for (auto & element : *this)
+            element.setting_name.clear(); /// We only make this element empty, the element will be removed in normalizeProfileElements().
+    }
+
+    auto apply_drop_setting = [&](const String & setting_name)
+    {
+        for (auto & element : *this)
+        {
+            if (element.setting_name == setting_name)
+                element.setting_name.clear();
+        }
+    };
+
+    auto apply_drop_profile = [&](const UUID & profile_id)
+    {
+        for (auto & element : *this)
+        {
+            if (element.parent_profile == profile_id)
+                element.parent_profile.reset();
+        }
+    };
+
+    for (const auto & drop : changes.drop_settings)
+    {
+        if (drop.parent_profile)
+            apply_drop_profile(*drop.parent_profile);
+        if (!drop.setting_name.empty())
+            apply_drop_setting(drop.setting_name);
+    }
+
+    auto apply_modify_setting = [&](const SettingsProfileElement & modify)
+    {
+        SettingsProfileElement new_element;
+        new_element.setting_name = modify.setting_name;
+        new_element.value = modify.value;
+        new_element.min_value = modify.min_value;
+        new_element.max_value = modify.max_value;
+        new_element.writability = modify.writability;
+        push_back(new_element); /// normalizeProfileElements() will merge this new element with the previous elements.
+    };
+
+    /// Apply "ADD" changes.
+    auto apply_add_setting = [&](const SettingsProfileElement & add)
+    {
+        /// "ADD SETTING" must replace the value and the constraints of a setting, so first we need drop the previous elements for that setting.
+        apply_drop_setting(add.setting_name);
+        apply_modify_setting(add);
+    };
+
+    auto apply_add_profile = [&](const UUID & profile_id)
+    {
+        SettingsProfileElement new_element;
+        new_element.parent_profile = profile_id;
+        push_back(new_element); /// We don't care about possible duplicates here, normalizeProfileElements() will remove duplicates.
+    };
+
+    for (const auto & add : changes.add_settings)
+    {
+        if (add.parent_profile)
+            apply_add_profile(*add.parent_profile);
+        if (!add.setting_name.empty())
+            apply_add_setting(add);
+    }
+
+    /// Apply "MODIFY" changes.
+    for (const auto & modify : changes.modify_settings)
+    {
+        chassert(!modify.parent_profile); /// There is no such thing as "MODIFY PROFILE".
+        if (!modify.setting_name.empty())
+            apply_modify_setting(modify);
+    }
+
+    /// Remove empty elements and duplicates, and sort the result.
+    normalize();
+}
+
 }

src/Access/SettingsProfileElement.h

@@ -13,8 +13,10 @@ namespace DB
 struct Settings;
 class SettingsChanges;
 class SettingsConstraints;
+struct AlterSettingsProfileElements;
 class ASTSettingsProfileElement;
 class ASTSettingsProfileElements;
+class ASTAlterSettingsProfileElements;
 class AccessControl;
 
 
@@ -44,6 +46,8 @@ struct SettingsProfileElement
     std::shared_ptr<ASTSettingsProfileElement> toAST() const;
     std::shared_ptr<ASTSettingsProfileElement> toASTWithNames(const AccessControl & access_control) const;
 
+    bool empty() const { return !parent_profile && (setting_name.empty() || (!value && !min_value && !max_value && !writability)); }
+
     bool isConstraint() const;
 
 private:
@@ -57,8 +61,9 @@ public:
     SettingsProfileElements() = default;
 
     /// The constructor from AST requires the AccessControl if `ast.id_mode == false`.
-    SettingsProfileElements(const ASTSettingsProfileElements & ast); /// NOLINT
-    SettingsProfileElements(const ASTSettingsProfileElements & ast, const AccessControl & access_control);
+    SettingsProfileElements(const ASTSettingsProfileElements & ast, bool normalize_ = true); /// NOLINT
+    SettingsProfileElements(const ASTSettingsProfileElements & ast, const AccessControl & access_control, bool normalize_ = true);
+
     std::shared_ptr<ASTSettingsProfileElements> toAST() const;
     std::shared_ptr<ASTSettingsProfileElements> toASTWithNames(const AccessControl & access_control) const;
 
@@ -70,16 +75,41 @@ public:
 
     void removeSettingsKeepProfiles();
 
-    void merge(const SettingsProfileElements & other);
-
     Settings toSettings() const;
     SettingsChanges toSettingsChanges() const;
     SettingsConstraints toSettingsConstraints(const AccessControl & access_control) const;
     std::vector<UUID> toProfileIDs() const;
 
-    bool isBackupAllowed() const;
+    /// Normalizes this list of profile elements: removes duplicates and empty elements, and also sorts the elements
+    /// in the following order: first profiles, then settings.
+    /// The function is called automatically after parsing profile elements from an AST and
+    /// at the end of an "ALTER PROFILE (USER/ROLE) command".
+    void normalize();
 
+    /// Appends all the elements of another list of profile elements to this list.
+    void merge(const SettingsProfileElements & other, bool normalize_ = true);
+
+    /// Applies changes from an "ALTER PROFILE (USER/ROLE)" command. Always normalizes the result.
+    void applyChanges(const AlterSettingsProfileElements & changes);
+
+    bool isBackupAllowed() const;
     static bool isAllowBackupSetting(const String & setting_name);
 };
 
+struct AlterSettingsProfileElements
+{
+    bool drop_all_settings = false;
+    bool drop_all_profiles = false;
+    SettingsProfileElements add_settings;
+    SettingsProfileElements modify_settings;
+    SettingsProfileElements drop_settings;
+
+    AlterSettingsProfileElements() = default;
+    explicit AlterSettingsProfileElements(const SettingsProfileElements & ast);
+    explicit AlterSettingsProfileElements(const ASTSettingsProfileElements & ast);
+    explicit AlterSettingsProfileElements(const ASTAlterSettingsProfileElements & ast);
+    AlterSettingsProfileElements(const ASTSettingsProfileElements & ast, const AccessControl & access_control);
+    AlterSettingsProfileElements(const ASTAlterSettingsProfileElements & ast, const AccessControl & access_control);
+};
+
 }

src/Access/SettingsProfilesCache.cpp

@@ -135,8 +135,8 @@ void SettingsProfilesCache::mergeSettingsAndConstraintsFor(EnabledSettings & ena
             merged_settings.emplace_back(new_element);
         }
 
-    merged_settings.merge(enabled.params.settings_from_enabled_roles);
-    merged_settings.merge(enabled.params.settings_from_user);
+    merged_settings.merge(enabled.params.settings_from_enabled_roles, /* normalize= */ false);
+    merged_settings.merge(enabled.params.settings_from_user, /* normalize= */ false);
 
     auto info = std::make_shared<SettingsProfilesInfo>(access_control);
 

src/Functions/array/arrayWithConstant.cpp

@@ -62,16 +62,17 @@ public:
         for (size_t i = 0; i < num_rows; ++i)
         {
             auto array_size = col_num->getInt(i);
+            auto element_size = col_value->byteSizeAt(i);
 
             if (unlikely(array_size < 0))
                 throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Array size {} cannot be negative: while executing function {}", array_size, getName());
 
             Int64 estimated_size = 0;
-            if (unlikely(common::mulOverflow(array_size, col_value->byteSize(), estimated_size)))
-                throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Array size {} with element size {} bytes is too large: while executing function {}", array_size, col_value->byteSize(), getName());
+            if (unlikely(common::mulOverflow(array_size, element_size, estimated_size)))
+                throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Array size {} with element size {} bytes is too large: while executing function {}", array_size, element_size, getName());
 
             if (unlikely(estimated_size > max_array_size_in_columns_bytes))
-                throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Array size {} with element size {} bytes is too large: while executing function {}", array_size, col_value->byteSize(), getName());
+                throw Exception(ErrorCodes::TOO_LARGE_ARRAY_SIZE, "Array size {} with element size {} bytes is too large: while executing function {}", array_size, element_size, getName());
 
             offset += array_size;
 

src/Interpreters/Access/InterpreterCreateRoleQuery.cpp

@@ -23,7 +23,7 @@ namespace
         Role & role,
         const ASTCreateRoleQuery & query,
         const String & override_name,
-        const std::optional<SettingsProfileElements> & override_settings)
+        const std::optional<AlterSettingsProfileElements> & override_settings)
     {
         if (!override_name.empty())
             role.setName(override_name);
@@ -33,9 +33,11 @@ namespace
             role.setName(query.names.front());
 
         if (override_settings)
-            role.settings = *override_settings;
+            role.settings.applyChanges(*override_settings);
+        else if (query.alter_settings)
+            role.settings.applyChanges(AlterSettingsProfileElements{*query.alter_settings});
         else if (query.settings)
-            role.settings = *query.settings;
+            role.settings.applyChanges(AlterSettingsProfileElements{*query.settings});
     }
 }
 
@@ -51,14 +53,14 @@ BlockIO InterpreterCreateRoleQuery::execute()
     else
         getContext()->checkAccess(AccessType::CREATE_ROLE);
 
-    std::optional<SettingsProfileElements> settings_from_query;
-    if (query.settings)
-    {
-        settings_from_query = SettingsProfileElements{*query.settings, access_control};
+    std::optional<AlterSettingsProfileElements> settings_from_query;
+    if (query.alter_settings)
+        settings_from_query = AlterSettingsProfileElements{*query.alter_settings, access_control};
+    else if (query.settings)
+        settings_from_query = AlterSettingsProfileElements{SettingsProfileElements(*query.settings, access_control)};
 
-        if (!query.attach)
-            getContext()->checkSettingsConstraints(*settings_from_query, SettingSource::ROLE);
-    }
+    if (settings_from_query && !query.attach)
+        getContext()->checkSettingsConstraints(*settings_from_query, SettingSource::ROLE);
 
     if (!query.cluster.empty())
         return executeDDLQueryOnCluster(updated_query_ptr, getContext());

src/Interpreters/Access/InterpreterCreateSettingsProfileQuery.cpp

@@ -25,7 +25,7 @@ namespace
         SettingsProfile & profile,
         const ASTCreateSettingsProfileQuery & query,
         const String & override_name,
-        const std::optional<SettingsProfileElements> & override_settings,
+        const std::optional<AlterSettingsProfileElements> & override_settings,
         const std::optional<RolesOrUsersSet> & override_to_roles)
     {
         if (!override_name.empty())
@@ -36,9 +36,11 @@ namespace
             profile.setName(query.names.front());
 
         if (override_settings)
-            profile.elements = *override_settings;
+            profile.elements.applyChanges(*override_settings);
+        else if (query.alter_settings)
+            profile.elements.applyChanges(AlterSettingsProfileElements{*query.alter_settings});
         else if (query.settings)
-            profile.elements = *query.settings;
+            profile.elements.applyChanges(AlterSettingsProfileElements{*query.settings});
 
         if (override_to_roles)
             profile.to_roles = *override_to_roles;
@@ -59,14 +61,14 @@ BlockIO InterpreterCreateSettingsProfileQuery::execute()
     else
         getContext()->checkAccess(AccessType::CREATE_SETTINGS_PROFILE);
 
-    std::optional<SettingsProfileElements> settings_from_query;
-    if (query.settings)
-    {
-        settings_from_query = SettingsProfileElements{*query.settings, access_control};
+    std::optional<AlterSettingsProfileElements> settings_from_query;
+    if (query.alter_settings)
+        settings_from_query = AlterSettingsProfileElements{*query.alter_settings, access_control};
+    else if (query.settings)
+        settings_from_query = AlterSettingsProfileElements{SettingsProfileElements(*query.settings, access_control)};
 
-        if (!query.attach)
-            getContext()->checkSettingsConstraints(*settings_from_query, SettingSource::PROFILE);
-    }
+    if (settings_from_query && !query.attach)
+        getContext()->checkSettingsConstraints(*settings_from_query, SettingSource::PROFILE);
 
     if (!query.cluster.empty())
     {

src/Interpreters/Access/InterpreterCreateUserQuery.cpp

@@ -43,7 +43,7 @@ namespace
         const std::vector<AuthenticationData> authentication_methods,
         const std::shared_ptr<ASTUserNameWithHost> & override_name,
         const std::optional<RolesOrUsersSet> & override_default_roles,
-        const std::optional<SettingsProfileElements> & override_settings,
+        const std::optional<AlterSettingsProfileElements> & override_settings,
         const std::optional<RolesOrUsersSet> & override_grantees,
         const std::optional<time_t> & global_valid_until,
         bool reset_authentication_methods,
@@ -172,9 +172,11 @@ namespace
             user.default_database = query.default_database->database_name;
 
         if (override_settings)
-            user.settings = *override_settings;
+            user.settings.applyChanges(*override_settings);
+        else if (query.alter_settings)
+            user.settings.applyChanges(AlterSettingsProfileElements{*query.alter_settings});
         else if (query.settings)
-            user.settings = *query.settings;
+            user.settings.applyChanges(AlterSettingsProfileElements{*query.settings});
 
         if (override_grantees)
             user.grantees = *override_grantees;
@@ -219,14 +221,14 @@ BlockIO InterpreterCreateUserQuery::execute()
         }
     }
 
-    std::optional<SettingsProfileElements> settings_from_query;
-    if (query.settings)
-    {
-        settings_from_query = SettingsProfileElements{*query.settings, access_control};
+    std::optional<AlterSettingsProfileElements> settings_from_query;
+    if (query.alter_settings)
+        settings_from_query = AlterSettingsProfileElements{*query.alter_settings, access_control};
+    else if (query.settings)
+        settings_from_query = AlterSettingsProfileElements{*query.settings, access_control};
 
-        if (!query.attach)
-            getContext()->checkSettingsConstraints(*settings_from_query, SettingSource::USER);
-    }
+    if (settings_from_query && !query.attach)
+        getContext()->checkSettingsConstraints(*settings_from_query, SettingSource::USER);
 
     if (!query.cluster.empty())
         return executeDDLQueryOnCluster(updated_query_ptr, getContext());

src/Interpreters/Access/InterpreterShowCreateAccessEntityQuery.cpp

@@ -71,10 +71,13 @@ namespace
 
         if (!user.settings.empty())
         {
+            std::shared_ptr<ASTSettingsProfileElements> query_settings;
             if (attach_mode)
-                query->settings = user.settings.toAST();
+                query_settings = user.settings.toAST();
             else
-                query->settings = user.settings.toASTWithNames(*access_control);
+                query_settings = user.settings.toASTWithNames(*access_control);
+            if (!query_settings->empty())
+                query->settings = query_settings;
         }
 
         if (user.grantees != RolesOrUsersSet::AllTag{})
@@ -105,10 +108,13 @@ namespace
 
         if (!role.settings.empty())
         {
+            std::shared_ptr<ASTSettingsProfileElements> query_settings;
             if (attach_mode)
-                query->settings = role.settings.toAST();
+                query_settings = role.settings.toAST();
             else
-                query->settings = role.settings.toASTWithNames(*access_control);
+                query_settings = role.settings.toASTWithNames(*access_control);
+            if (!query_settings->empty())
+                query->settings = query_settings;
         }
 
         return query;
@@ -123,12 +129,16 @@ namespace
 
         if (!profile.elements.empty())
         {
+            std::shared_ptr<ASTSettingsProfileElements> query_settings;
             if (attach_mode)
-                query->settings = profile.elements.toAST();
+                query_settings = profile.elements.toAST();
             else
-                query->settings = profile.elements.toASTWithNames(*access_control);
-            if (query->settings)
-                query->settings->setUseInheritKeyword(true);
+                query_settings = profile.elements.toASTWithNames(*access_control);
+            if (!query_settings->empty())
+            {
+                query_settings->setUseInheritKeyword(true);
+                query->settings = query_settings;
+            }
         }
 
         if (!profile.to_roles.empty())

src/Interpreters/Context.cpp

@@ -2521,7 +2521,7 @@ void Context::applySettingsChanges(const SettingsChanges & changes)
     applySettingsChangesWithLock(changes, lock);
 }
 
-void Context::checkSettingsConstraintsWithLock(const SettingsProfileElements & profile_elements, SettingSource source)
+void Context::checkSettingsConstraintsWithLock(const AlterSettingsProfileElements & profile_elements, SettingSource source)
 {
     getSettingsConstraintsAndCurrentProfilesWithLock()->constraints.check(*settings, profile_elements, source);
     if (getApplicationType() == ApplicationType::LOCAL || getApplicationType() == ApplicationType::SERVER)
@@ -2561,7 +2561,7 @@ void Context::checkMergeTreeSettingsConstraintsWithLock(const MergeTreeSettings
     getSettingsConstraintsAndCurrentProfilesWithLock()->constraints.check(merge_tree_settings, changes);
 }
 
-void Context::checkSettingsConstraints(const SettingsProfileElements & profile_elements, SettingSource source)
+void Context::checkSettingsConstraints(const AlterSettingsProfileElements & profile_elements, SettingSource source)
 {
     SharedLockGuard lock(mutex);
     checkSettingsConstraintsWithLock(profile_elements, source);

src/Interpreters/Context.h

@@ -145,7 +145,7 @@ struct Settings;
 struct SettingChange;
 class SettingsChanges;
 struct SettingsConstraintsAndProfileIDs;
-class SettingsProfileElements;
+struct AlterSettingsProfileElements;
 class RemoteHostFilter;
 class IDisk;
 using DiskPtr = std::shared_ptr<IDisk>;
@@ -865,7 +865,7 @@ public:
     void applySettingsChanges(const SettingsChanges & changes);
 
     /// Checks the constraints.
-    void checkSettingsConstraints(const SettingsProfileElements & profile_elements, SettingSource source);
+    void checkSettingsConstraints(const AlterSettingsProfileElements & profile_elements, SettingSource source);
     void checkSettingsConstraints(const SettingChange & change, SettingSource source);
     void checkSettingsConstraints(const SettingsChanges & changes, SettingSource source);
     void checkSettingsConstraints(SettingsChanges & changes, SettingSource source);
@@ -1413,7 +1413,7 @@ private:
 
     void setCurrentDatabaseWithLock(const String & name, const std::lock_guard<ContextSharedMutex> & lock);
 
-    void checkSettingsConstraintsWithLock(const SettingsProfileElements & profile_elements, SettingSource source);
+    void checkSettingsConstraintsWithLock(const AlterSettingsProfileElements & profile_elements, SettingSource source);
 
     void checkSettingsConstraintsWithLock(const SettingChange & change, SettingSource source);
 
@@ -1422,6 +1422,9 @@ private:
     void checkSettingsConstraintsWithLock(SettingsChanges & changes, SettingSource source);
 
     void clampToSettingsConstraintsWithLock(SettingsChanges & changes, SettingSource source);
+    void checkSettingsConstraintsWithLock(const AlterSettingsProfileElements & profile_elements, SettingSource source) const;
+
+    void clampToSettingsConstraintsWithLock(SettingsChanges & changes, SettingSource source) const;
 
     void checkMergeTreeSettingsConstraintsWithLock(const MergeTreeSettings & merge_tree_settings, const SettingsChanges & changes) const;
 

src/Parsers/Access/ASTCreateRoleQuery.cpp

@@ -31,6 +31,12 @@ namespace
         format.ostr << (format.hilite ? IAST::hilite_keyword : "") << " SETTINGS " << (format.hilite ? IAST::hilite_none : "");
         settings.format(format);
     }
+
+    void formatAlterSettings(const ASTAlterSettingsProfileElements & alter_settings, const IAST::FormatSettings & format)
+    {
+        format.ostr << " ";
+        alter_settings.format(format);
+    }
 }
 
 
@@ -47,6 +53,9 @@ ASTPtr ASTCreateRoleQuery::clone() const
     if (settings)
         res->settings = std::static_pointer_cast<ASTSettingsProfileElements>(settings->clone());
 
+    if (alter_settings)
+        res->alter_settings = std::static_pointer_cast<ASTAlterSettingsProfileElements>(alter_settings->clone());
+
     return res;
 }
 
@@ -82,7 +91,9 @@ void ASTCreateRoleQuery::formatImpl(const FormatSettings & format, FormatState &
     if (!new_name.empty())
         formatRenameTo(new_name, format);
 
-    if (settings && (!settings->empty() || alter))
+    if (alter_settings)
+        formatAlterSettings(*alter_settings, format);
+    else if (settings)
         formatSettings(*settings, format);
 }
 

src/Parsers/Access/ASTCreateRoleQuery.h

@@ -7,6 +7,7 @@
 namespace DB
 {
 class ASTSettingsProfileElements;
+class ASTAlterSettingsProfileElements;
 
 
 /** CREATE ROLE [IF NOT EXISTS | OR REPLACE] name
@@ -14,7 +15,12 @@ class ASTSettingsProfileElements;
   *
   * ALTER ROLE [IF EXISTS] name
   *     [RENAME TO new_name]
-  *     [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+  *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+  *     [DROP SETTINGS variable [,...] ]
+  *     [ADD PROFILES 'profile_name' [,...] ]
+  *     [DROP PROFILES 'profile_name' [,...] ]
+  *     [DROP ALL PROFILES]
+  *     [DROP ALL SETTINGS]
   */
 class ASTCreateRoleQuery : public IAST, public ASTQueryWithOnCluster
 {
@@ -31,6 +37,7 @@ public:
     String storage_name;
 
     std::shared_ptr<ASTSettingsProfileElements> settings;
+    std::shared_ptr<ASTAlterSettingsProfileElements> alter_settings;
 
     String getID(char) const override;
     ASTPtr clone() const override;

src/Parsers/Access/ASTCreateSettingsProfileQuery.cpp

@@ -33,6 +33,12 @@ namespace
         settings.format(format);
     }
 
+    void formatAlterSettings(const ASTAlterSettingsProfileElements & alter_settings, const IAST::FormatSettings & format)
+    {
+        format.ostr << " ";
+        alter_settings.format(format);
+    }
+
     void formatToRoles(const ASTRolesOrUsersSet & roles, const IAST::FormatSettings & settings)
     {
         settings.ostr << (settings.hilite ? IAST::hilite_keyword : "") << " TO " << (settings.hilite ? IAST::hilite_none : "");
@@ -57,6 +63,9 @@ ASTPtr ASTCreateSettingsProfileQuery::clone() const
     if (settings)
         res->settings = std::static_pointer_cast<ASTSettingsProfileElements>(settings->clone());
 
+    if (alter_settings)
+        res->alter_settings = std::static_pointer_cast<ASTAlterSettingsProfileElements>(alter_settings->clone());
+
     return res;
 }
 
@@ -92,7 +101,9 @@ void ASTCreateSettingsProfileQuery::formatImpl(const FormatSettings & format, Fo
     if (!new_name.empty())
         formatRenameTo(new_name, format);
 
-    if (settings && (!settings->empty() || alter))
+    if (alter_settings)
+        formatAlterSettings(*alter_settings, format);
+    else if (settings)
         formatSettings(*settings, format);
 
     if (to_roles && (!to_roles->empty() || alter))

src/Parsers/Access/ASTCreateSettingsProfileQuery.h

@@ -7,6 +7,7 @@
 namespace DB
 {
 class ASTSettingsProfileElements;
+class ASTAlterSettingsProfileElements;
 class ASTRolesOrUsersSet;
 
 
@@ -16,7 +17,12 @@ class ASTRolesOrUsersSet;
   *
   * ALTER SETTINGS PROFILE [IF EXISTS] name
   *     [RENAME TO new_name]
-  *     [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+  *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+  *     [DROP SETTINGS variable [,...] ]
+  *     [ADD PROFILES 'profile_name' [,...] ]
+  *     [DROP PROFILES 'profile_name' [,...] ]
+  *     [DROP ALL PROFILES]
+  *     [DROP ALL SETTINGS]
   *     [TO {role [,...] | ALL | ALL EXCEPT role [,...]}]
   */
 class ASTCreateSettingsProfileQuery : public IAST, public ASTQueryWithOnCluster
@@ -34,6 +40,7 @@ public:
     String new_name;
 
     std::shared_ptr<ASTSettingsProfileElements> settings;
+    std::shared_ptr<ASTAlterSettingsProfileElements> alter_settings;
 
     std::shared_ptr<ASTRolesOrUsersSet> to_roles;
 
@@ -44,4 +51,5 @@ public:
     ASTPtr getRewrittenASTWithoutOnCluster(const WithoutOnClusterASTRewriteParams &) const override { return removeOnCluster<ASTCreateSettingsProfileQuery>(clone()); }
     QueryKind getQueryKind() const override { return QueryKind::Create; }
 };
+
 }

src/Parsers/Access/ASTCreateUserQuery.cpp

@@ -149,13 +149,17 @@ namespace
         default_roles.format(settings);
     }
 
-
     void formatSettings(const ASTSettingsProfileElements & settings, const IAST::FormatSettings & format)
     {
         format.ostr << (format.hilite ? IAST::hilite_keyword : "") << " SETTINGS " << (format.hilite ? IAST::hilite_none : "");
         settings.format(format);
     }
 
+    void formatAlterSettings(const ASTAlterSettingsProfileElements & alter_settings, const IAST::FormatSettings & format)
+    {
+        format.ostr << " ";
+        alter_settings.format(format);
+    }
 
     void formatGrantees(const ASTRolesOrUsersSet & grantees, const IAST::FormatSettings & settings)
     {
@@ -198,6 +202,9 @@ ASTPtr ASTCreateUserQuery::clone() const
     if (settings)
         res->settings = std::static_pointer_cast<ASTSettingsProfileElements>(settings->clone());
 
+    if (alter_settings)
+        res->alter_settings = std::static_pointer_cast<ASTAlterSettingsProfileElements>(alter_settings->clone());
+
     for (const auto & authentication_method : authentication_methods)
     {
         auto ast_clone = std::static_pointer_cast<ASTAuthenticationData>(authentication_method->clone());
@@ -278,7 +285,9 @@ void ASTCreateUserQuery::formatImpl(const FormatSettings & format, FormatState &
     if (default_roles)
         formatDefaultRoles(*default_roles, format);
 
-    if (settings && (!settings->empty() || alter))
+    if (alter_settings)
+        formatAlterSettings(*alter_settings, format);
+    else if (settings)
         formatSettings(*settings, format);
 
     if (grantees)

src/Parsers/Access/ASTCreateUserQuery.h

@@ -13,6 +13,7 @@ class ASTUserNamesWithHost;
 class ASTRolesOrUsersSet;
 class ASTDatabaseOrNone;
 class ASTSettingsProfileElements;
+class ASTAlterSettingsProfileElements;
 class ASTAuthenticationData;
 
 
@@ -30,7 +31,12 @@ class ASTAuthenticationData;
   *     [[ADD|DROP] HOST {LOCAL | NAME 'name' | REGEXP 'name_regexp' | IP 'address' | LIKE 'pattern'} [,...] | ANY | NONE]
   *     [DEFAULT ROLE role [,...] | ALL | ALL EXCEPT role [,...] ]
   *     [DEFAULT DATABASE database | NONE]
-  *     [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+  *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+  *     [DROP SETTINGS variable [,...] ]
+  *     [ADD PROFILES 'profile_name' [,...] ]
+  *     [DROP PROFILES 'profile_name' [,...] ]
+  *     [DROP ALL PROFILES]
+  *     [DROP ALL SETTINGS]
   *     [GRANTEES {user | role | ANY | NONE} [,...] [EXCEPT {user | role} [,...]]]
   */
 class ASTCreateUserQuery : public IAST, public ASTQueryWithOnCluster
@@ -58,6 +64,7 @@ public:
 
     std::shared_ptr<ASTRolesOrUsersSet> default_roles;
     std::shared_ptr<ASTSettingsProfileElements> settings;
+    std::shared_ptr<ASTAlterSettingsProfileElements> alter_settings;
     std::shared_ptr<ASTRolesOrUsersSet> grantees;
 
     std::shared_ptr<ASTDatabaseOrNone> default_database;

src/Parsers/Access/ASTSettingsProfileElement.cpp

@@ -3,6 +3,7 @@
 #include <Common/FieldVisitorToString.h>
 #include <Common/quoteString.h>
 #include <IO/Operators.h>
+#include <base/insertAtEnd.h>
 
 
 namespace DB
@@ -21,8 +22,54 @@ namespace
             settings.ostr << backQuote(str);
         }
     }
+
+    void formatSettingsProfileElementsForAlter(std::string_view kind, const ASTSettingsProfileElements & elements, const IAST::FormatSettings & settings)
+    {
+        bool need_comma = false;
+
+        size_t num_profiles = elements.getNumberOfProfiles();
+        if (num_profiles > 0)
+        {
+            settings.ostr << (settings.hilite ? IAST::hilite_keyword : "") << kind << " " << (num_profiles == 1 ? "PROFILE" : "PROFILES")
+                          << (settings.hilite ? IAST::hilite_none : "") << " ";
+
+            for (const auto & element : elements.elements)
+            {
+                if (!element->parent_profile.empty())
+                {
+                    if (need_comma)
+                        settings.ostr << ", ";
+                    formatProfileNameOrID(element->parent_profile, /* is_id= */ false, settings);
+                    need_comma = true;
+                }
+            }
+        }
+
+        size_t num_settings = elements.getNumberOfSettings();
+        if (num_settings > 0)
+        {
+            if (need_comma)
+                settings.ostr << ", ";
+            need_comma = false;
+
+            settings.ostr << (settings.hilite ? IAST::hilite_keyword : "") << kind << " " << (num_settings == 1 ? "SETTING" : "SETTINGS")
+                          << (settings.hilite ? IAST::hilite_none : "") << " ";
+
+            for (const auto & element : elements.elements)
+            {
+                if (!element->setting_name.empty())
+                {
+                    if (need_comma)
+                        settings.ostr << ", ";
+                    element->format(settings);
+                    need_comma = true;
+                }
+            }
+        }
+    }
 }
 
+
 void ASTSettingsProfileElement::formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const
 {
     if (!parent_profile.empty())
@@ -82,6 +129,27 @@ bool ASTSettingsProfileElements::empty() const
     return true;
 }
 
+size_t ASTSettingsProfileElements::getNumberOfSettings() const
+{
+    return std::count_if(elements.begin(), elements.end(), [](const auto & element){ return !element->setting_name.empty(); });
+}
+
+size_t ASTSettingsProfileElements::getNumberOfProfiles() const
+{
+    return std::count_if(elements.begin(), elements.end(), [](const auto & element){ return !element->parent_profile.empty(); });
+}
+
+
+ASTPtr ASTSettingsProfileElements::clone() const
+{
+    auto res = std::make_shared<ASTSettingsProfileElements>(*this);
+
+    for (auto & element : res->elements)
+        element = std::static_pointer_cast<ASTSettingsProfileElement>(element->clone());
+
+    return res;
+}
+
 
 void ASTSettingsProfileElements::formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const
 {
@@ -109,4 +177,108 @@ void ASTSettingsProfileElements::setUseInheritKeyword(bool use_inherit_keyword_)
         element->use_inherit_keyword = use_inherit_keyword_;
 }
 
+
+void ASTSettingsProfileElements::add(ASTSettingsProfileElements && other)
+{
+    insertAtEnd(elements, std::move(other.elements));
+}
+
+
+String ASTAlterSettingsProfileElements::getID(char) const
+{
+    return "AlterSettingsProfileElements";
+}
+
+ASTPtr ASTAlterSettingsProfileElements::clone() const
+{
+    auto res = std::make_shared<ASTAlterSettingsProfileElements>(*this);
+
+    if (add_settings)
+        res->add_settings = std::static_pointer_cast<ASTSettingsProfileElements>(add_settings->clone());
+
+    if (modify_settings)
+        res->modify_settings = std::static_pointer_cast<ASTSettingsProfileElements>(modify_settings->clone());
+
+    if (drop_settings)
+        res->drop_settings = std::static_pointer_cast<ASTSettingsProfileElements>(drop_settings->clone());
+
+    return res;
+}
+
+void ASTAlterSettingsProfileElements::formatImpl(const FormatSettings & format, FormatState &, FormatStateStacked) const
+{
+    bool need_comma = false;
+
+    if (drop_all_settings)
+    {
+        format.ostr << (format.hilite ? IAST::hilite_keyword : "") << "DROP ALL SETTINGS" << (format.hilite ? IAST::hilite_none : "");
+        need_comma = true;
+    }
+
+    if (drop_all_profiles)
+    {
+        if (need_comma)
+            format.ostr << ", ";
+        format.ostr << (format.hilite ? IAST::hilite_keyword : "") << "DROP ALL PROFILES" << (format.hilite ? IAST::hilite_none : "");
+        need_comma = true;
+    }
+
+    if (drop_settings && !drop_settings->empty())
+    {
+        if (need_comma)
+            format.ostr << ", ";
+        formatSettingsProfileElementsForAlter("DROP", *drop_settings, format);
+        need_comma = true;
+    }
+
+    if (add_settings && !add_settings->empty())
+    {
+        if (need_comma)
+            format.ostr << ", ";
+        formatSettingsProfileElementsForAlter("ADD", *add_settings, format);
+        need_comma = true;
+    }
+
+    if (modify_settings && !modify_settings->empty())
+    {
+        if (need_comma)
+            format.ostr << ", ";
+        formatSettingsProfileElementsForAlter("MODIFY", *modify_settings, format);
+    }
+}
+
+void ASTAlterSettingsProfileElements::add(ASTAlterSettingsProfileElements && other)
+{
+    drop_all_settings |= other.drop_all_settings;
+    drop_all_profiles |= other.drop_all_profiles;
+
+    if (other.add_settings)
+    {
+        if (!add_settings)
+            add_settings = std::make_shared<ASTSettingsProfileElements>();
+        add_settings->add(std::move(*other.add_settings));
+    }
+
+    if (other.add_settings)
+    {
+        if (!add_settings)
+            add_settings = std::make_shared<ASTSettingsProfileElements>();
+        add_settings->add(std::move(*other.add_settings));
+    }
+
+    if (other.modify_settings)
+    {
+        if (!modify_settings)
+            modify_settings = std::make_shared<ASTSettingsProfileElements>();
+        modify_settings->add(std::move(*other.modify_settings));
+    }
+
+    if (other.drop_settings)
+    {
+        if (!drop_settings)
+            drop_settings = std::make_shared<ASTSettingsProfileElements>();
+        drop_settings->add(std::move(*other.drop_settings));
+    }
+}
+
 }

src/Parsers/Access/ASTSettingsProfileElement.h

@@ -39,10 +39,41 @@ public:
 
     bool empty() const;
 
+    size_t getNumberOfSettings() const;
+    size_t getNumberOfProfiles() const;
+
     String getID(char) const override { return "SettingsProfileElements"; }
-    ASTPtr clone() const override { return std::make_shared<ASTSettingsProfileElements>(*this); }
+    ASTPtr clone() const override;
     void formatImpl(const FormatSettings & settings, FormatState &, FormatStateStacked) const override;
 
     void setUseInheritKeyword(bool use_inherit_keyword_);
+
+    void add(ASTSettingsProfileElements && other);
 };
+
+/* Represents a clause used to alter settings or profiles assigned to a user or a role or another profile.
+ *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+ *     [DROP SETTINGS variable [,...] ]
+ *     [ADD PROFILES 'profile_name' [,...] ]
+ *     [DROP PROFILES 'profile_name' [,...] ]
+ *     [DROP ALL PROFILES]
+ *     [DROP ALL SETTINGS]
+*/
+class ASTAlterSettingsProfileElements : public IAST
+{
+public:
+    std::shared_ptr<ASTSettingsProfileElements> add_settings;
+    std::shared_ptr<ASTSettingsProfileElements> modify_settings;
+    std::shared_ptr<ASTSettingsProfileElements> drop_settings;
+
+    bool drop_all_settings = false;
+    bool drop_all_profiles = false;
+
+    String getID(char) const override;
+    ASTPtr clone() const override;
+    void formatImpl(const FormatSettings & format, FormatState &, FormatStateStacked) const override;
+
+    void add(ASTAlterSettingsProfileElements && other);
+};
+
 }

src/Parsers/Access/ParserCreateRoleQuery.cpp

@@ -25,20 +25,31 @@ namespace
         });
     }
 
-    bool parseSettings(IParserBase::Pos & pos, Expected & expected, bool id_mode, std::vector<std::shared_ptr<ASTSettingsProfileElement>> & settings)
+    bool parseSettings(IParserBase::Pos & pos, Expected & expected, bool id_mode, std::shared_ptr<ASTSettingsProfileElements> & settings)
     {
         return IParserBase::wrapParseImpl(pos, [&]
         {
-            if (!ParserKeyword{Keyword::SETTINGS}.ignore(pos, expected))
-                return false;
-
-            ASTPtr new_settings_ast;
+            ASTPtr ast;
             ParserSettingsProfileElements elements_p;
             elements_p.useIDMode(id_mode);
-            if (!elements_p.parse(pos, new_settings_ast, expected))
+            if (!elements_p.parse(pos, ast, expected))
                 return false;
 
-            settings = std::move(new_settings_ast->as<ASTSettingsProfileElements &>().elements);
+            settings = typeid_cast<std::shared_ptr<ASTSettingsProfileElements>>(ast);
+            return true;
+        });
+    }
+
+    bool parseAlterSettings(IParserBase::Pos & pos, Expected & expected, std::shared_ptr<ASTAlterSettingsProfileElements> & alter_settings)
+    {
+        return IParserBase::wrapParseImpl(pos, [&]
+        {
+            ASTPtr ast;
+            ParserAlterSettingsProfileElements elements_p;
+            if (!elements_p.parse(pos, ast, expected))
+                return false;
+
+            alter_settings = typeid_cast<std::shared_ptr<ASTAlterSettingsProfileElements>>(ast);
             return true;
         });
     }
@@ -91,6 +102,7 @@ bool ParserCreateRoleQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
 
     String new_name;
     std::shared_ptr<ASTSettingsProfileElements> settings;
+    std::shared_ptr<ASTAlterSettingsProfileElements> alter_settings;
     String cluster;
     String storage_name;
 
@@ -99,14 +111,27 @@ bool ParserCreateRoleQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
         if (alter && new_name.empty() && (names.size() == 1) && parseRenameTo(pos, expected, new_name))
             continue;
 
-        std::vector<std::shared_ptr<ASTSettingsProfileElement>> new_settings;
-        if (parseSettings(pos, expected, attach_mode, new_settings))
+        if (alter)
         {
-            if (!settings)
-                settings = std::make_shared<ASTSettingsProfileElements>();
-
-            insertAtEnd(settings->elements, std::move(new_settings));
-            continue;
+            std::shared_ptr<ASTAlterSettingsProfileElements> new_alter_settings;
+            if (parseAlterSettings(pos, expected, new_alter_settings))
+            {
+                if (!alter_settings)
+                    alter_settings = std::make_shared<ASTAlterSettingsProfileElements>();
+                alter_settings->add(std::move(*new_alter_settings));
+                continue;
+            }
+        }
+        else
+        {
+            std::shared_ptr<ASTSettingsProfileElements> new_settings;
+            if (parseSettings(pos, expected, attach_mode, new_settings))
+            {
+                if (!settings)
+                    settings = std::make_shared<ASTSettingsProfileElements>();
+                settings->add(std::move(*new_settings));
+                continue;
+            }
         }
 
         if (cluster.empty() && parseOnCluster(pos, expected, cluster))
@@ -130,6 +155,7 @@ bool ParserCreateRoleQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
     query->names = std::move(names);
     query->new_name = std::move(new_name);
     query->settings = std::move(settings);
+    query->alter_settings = std::move(alter_settings);
     query->storage_name = std::move(storage_name);
 
     return true;

src/Parsers/Access/ParserCreateRoleQuery.h

@@ -11,7 +11,12 @@ namespace DB
   *
   * ALTER ROLE [IF EXISTS] name
   *     [RENAME TO new_name]
-  *     [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+  *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+  *     [DROP SETTINGS variable [,...] ]
+  *     [ADD PROFILES 'profile_name' [,...] ]
+  *     [DROP PROFILES 'profile_name' [,...] ]
+  *     [DROP ALL PROFILES]
+  *     [DROP ALL SETTINGS]
   */
 class ParserCreateRoleQuery : public IParserBase
 {

src/Parsers/Access/ParserCreateSettingsProfileQuery.cpp

@@ -27,20 +27,32 @@ namespace
         });
     }
 
-    bool parseSettings(IParserBase::Pos & pos, Expected & expected, bool id_mode, std::vector<std::shared_ptr<ASTSettingsProfileElement>> & settings)
+    bool parseSettings(IParserBase::Pos & pos, Expected & expected, bool id_mode, std::shared_ptr<ASTSettingsProfileElements> & settings)
     {
         return IParserBase::wrapParseImpl(pos, [&]
         {
-            if (!ParserKeyword{Keyword::SETTINGS}.ignore(pos, expected))
-                return false;
-
-            ASTPtr new_settings_ast;
+            ASTPtr ast;
             ParserSettingsProfileElements elements_p;
             elements_p.useInheritKeyword(true).useIDMode(id_mode);
-            if (!elements_p.parse(pos, new_settings_ast, expected))
+            if (!elements_p.parse(pos, ast, expected))
                 return false;
 
-            settings = std::move(new_settings_ast->as<ASTSettingsProfileElements &>().elements);
+            settings = typeid_cast<std::shared_ptr<ASTSettingsProfileElements>>(ast);
+            return true;
+        });
+    }
+
+    bool parseAlterSettings(IParserBase::Pos & pos, Expected & expected, std::shared_ptr<ASTAlterSettingsProfileElements> & alter_settings)
+    {
+        return IParserBase::wrapParseImpl(pos, [&]
+        {
+            ASTPtr ast;
+            ParserAlterSettingsProfileElements elements_p;
+            elements_p.useInheritKeyword(true);
+            if (!elements_p.parse(pos, ast, expected))
+                return false;
+
+            alter_settings = typeid_cast<std::shared_ptr<ASTAlterSettingsProfileElements>>(ast);
             return true;
         });
     }
@@ -111,6 +123,7 @@ bool ParserCreateSettingsProfileQuery::parseImpl(Pos & pos, ASTPtr & node, Expec
 
     String new_name;
     std::shared_ptr<ASTSettingsProfileElements> settings;
+    std::shared_ptr<ASTAlterSettingsProfileElements> alter_settings;
     String cluster;
     String storage_name;
 
@@ -119,14 +132,27 @@ bool ParserCreateSettingsProfileQuery::parseImpl(Pos & pos, ASTPtr & node, Expec
         if (alter && new_name.empty() && (names.size() == 1) && parseRenameTo(pos, expected, new_name))
             continue;
 
-        std::vector<std::shared_ptr<ASTSettingsProfileElement>> new_settings;
-        if (parseSettings(pos, expected, attach_mode, new_settings))
+        if (alter)
         {
-            if (!settings)
-                settings = std::make_shared<ASTSettingsProfileElements>();
-
-            insertAtEnd(settings->elements, std::move(new_settings));
-            continue;
+            std::shared_ptr<ASTAlterSettingsProfileElements> new_alter_settings;
+            if (parseAlterSettings(pos, expected, new_alter_settings))
+            {
+                if (!alter_settings)
+                    alter_settings = std::make_shared<ASTAlterSettingsProfileElements>();
+                alter_settings->add(std::move(*new_alter_settings));
+                continue;
+            }
+        }
+        else
+        {
+            std::shared_ptr<ASTSettingsProfileElements> new_settings;
+            if (parseSettings(pos, expected, attach_mode, new_settings))
+            {
+                if (!settings)
+                    settings = std::make_shared<ASTSettingsProfileElements>();
+                settings->add(std::move(*new_settings));
+                continue;
+            }
         }
 
         if (cluster.empty() && parseOnCluster(pos, expected, cluster))
@@ -156,6 +182,7 @@ bool ParserCreateSettingsProfileQuery::parseImpl(Pos & pos, ASTPtr & node, Expec
     query->names = std::move(names);
     query->new_name = std::move(new_name);
     query->settings = std::move(settings);
+    query->alter_settings = std::move(alter_settings);
     query->to_roles = std::move(to_roles);
     query->storage_name = std::move(storage_name);
 

src/Parsers/Access/ParserCreateSettingsProfileQuery.h

@@ -11,7 +11,12 @@ namespace DB
   *
   * ALTER SETTINGS PROFILE [IF EXISTS] name
   *     [RENAME TO new_name]
-  *     [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | INHERIT 'profile_name'] [,...]
+  *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+  *     [DROP SETTINGS variable [,...] ]
+  *     [ADD PROFILES 'profile_name' [,...] ]
+  *     [DROP PROFILES 'profile_name' [,...] ]
+  *     [DROP ALL PROFILES]
+  *     [DROP ALL SETTINGS]
   */
 class ParserCreateSettingsProfileQuery : public IParserBase
 {

src/Parsers/Access/ParserCreateUserQuery.cpp

@@ -427,20 +427,31 @@ namespace
     }
 
 
-    bool parseSettings(IParserBase::Pos & pos, Expected & expected, bool id_mode, std::vector<std::shared_ptr<ASTSettingsProfileElement>> & settings)
+    bool parseSettings(IParserBase::Pos & pos, Expected & expected, bool id_mode, std::shared_ptr<ASTSettingsProfileElements> & settings)
     {
         return IParserBase::wrapParseImpl(pos, [&]
         {
-            if (!ParserKeyword{Keyword::SETTINGS}.ignore(pos, expected))
-                return false;
-
-            ASTPtr new_settings_ast;
+            ASTPtr ast;
             ParserSettingsProfileElements elements_p;
             elements_p.useIDMode(id_mode);
-            if (!elements_p.parse(pos, new_settings_ast, expected))
+            if (!elements_p.parse(pos, ast, expected))
                 return false;
 
-            settings = std::move(new_settings_ast->as<ASTSettingsProfileElements &>().elements);
+            settings = typeid_cast<std::shared_ptr<ASTSettingsProfileElements>>(ast);
+            return true;
+        });
+    }
+
+    bool parseAlterSettings(IParserBase::Pos & pos, Expected & expected, std::shared_ptr<ASTAlterSettingsProfileElements> & alter_settings)
+    {
+        return IParserBase::wrapParseImpl(pos, [&]
+        {
+            ASTPtr ast;
+            ParserAlterSettingsProfileElements elements_p;
+            if (!elements_p.parse(pos, ast, expected))
+                return false;
+
+            alter_settings = typeid_cast<std::shared_ptr<ASTAlterSettingsProfileElements>>(ast);
             return true;
         });
     }
@@ -556,6 +567,7 @@ bool ParserCreateUserQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
     std::vector<std::shared_ptr<ASTAuthenticationData>> auth_data;
     std::shared_ptr<ASTRolesOrUsersSet> default_roles;
     std::shared_ptr<ASTSettingsProfileElements> settings;
+    std::shared_ptr<ASTAlterSettingsProfileElements> alter_settings;
     std::shared_ptr<ASTRolesOrUsersSet> grantees;
     std::shared_ptr<ASTDatabaseOrNone> default_database;
     ASTPtr global_valid_until;
@@ -604,14 +616,27 @@ bool ParserCreateUserQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
             continue;
         }
 
-        std::vector<std::shared_ptr<ASTSettingsProfileElement>> new_settings;
-        if (parseSettings(pos, expected, attach_mode, new_settings))
+        if (alter)
         {
-            if (!settings)
-                settings = std::make_shared<ASTSettingsProfileElements>();
-
-            insertAtEnd(settings->elements, std::move(new_settings));
-            continue;
+            std::shared_ptr<ASTAlterSettingsProfileElements> new_alter_settings;
+            if (parseAlterSettings(pos, expected, new_alter_settings))
+            {
+                if (!alter_settings)
+                    alter_settings = std::make_shared<ASTAlterSettingsProfileElements>();
+                alter_settings->add(std::move(*new_alter_settings));
+                continue;
+            }
+        }
+        else
+        {
+            std::shared_ptr<ASTSettingsProfileElements> new_settings;
+            if (parseSettings(pos, expected, attach_mode, new_settings))
+            {
+                if (!settings)
+                    settings = std::make_shared<ASTSettingsProfileElements>();
+                settings->add(std::move(*new_settings));
+                continue;
+            }
         }
 
         if (!default_roles && parseDefaultRoles(pos, expected, attach_mode, default_roles))
@@ -691,6 +716,7 @@ bool ParserCreateUserQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expec
     query->remove_hosts = std::move(remove_hosts);
     query->default_roles = std::move(default_roles);
     query->settings = std::move(settings);
+    query->alter_settings = std::move(alter_settings);
     query->grantees = std::move(grantees);
     query->default_database = std::move(default_database);
     query->global_valid_until = std::move(global_valid_until);

src/Parsers/Access/ParserCreateUserQuery.h

@@ -18,7 +18,12 @@ namespace DB
   *     [NOT IDENTIFIED | IDENTIFIED {[WITH {no_password|plaintext_password|sha256_password|sha256_hash|double_sha1_password|double_sha1_hash}] BY {'password'|'hash'}}|{WITH ldap SERVER 'server_name'}|{WITH kerberos [REALM 'realm']}]
   *     [[ADD|DROP] HOST {LOCAL | NAME 'name' | REGEXP 'name_regexp' | IP 'address' | LIKE 'pattern'} [,...] | ANY | NONE]
   *     [DEFAULT ROLE role [,...] | ALL | ALL EXCEPT role [,...] ]
-  *     [SETTINGS variable [= value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] | PROFILE 'profile_name'] [,...]
+  *     [ADD|MODIFY SETTINGS variable [=value] [MIN [=] min_value] [MAX [=] max_value] [CONST|READONLY|WRITABLE|CHANGEABLE_IN_READONLY] [,...] ]
+  *     [DROP SETTINGS variable [,...] ]
+  *     [ADD PROFILES 'profile_name' [,...] ]
+  *     [DROP PROFILES 'profile_name' [,...] ]
+  *     [DROP ALL PROFILES]
+  *     [DROP ALL SETTINGS]
   *     [GRANTEES {user | role | ANY | NONE} [,...] [EXCEPT {user | role} [,...]]]
   */
 class ParserCreateUserQuery : public IParserBase

src/Parsers/Access/ParserSettingsProfileElement.cpp

@@ -7,27 +7,13 @@
 #include <Parsers/ExpressionListParsers.h>
 #include <Parsers/parseIdentifierOrStringLiteral.h>
 #include <boost/algorithm/string/predicate.hpp>
+#include <base/insertAtEnd.h>
 
 
 namespace DB
 {
 namespace
 {
-    bool parseProfileKeyword(IParserBase::Pos & pos, Expected & expected, bool use_inherit_keyword)
-    {
-        if (ParserKeyword{Keyword::PROFILE}.ignore(pos, expected))
-            return true;
-
-        if (use_inherit_keyword && ParserKeyword{Keyword::INHERIT}.ignore(pos, expected))
-        {
-            ParserKeyword{Keyword::PROFILE}.ignore(pos, expected);
-            return true;
-        }
-
-        return false;
-    }
-
-
     bool parseProfileNameOrID(IParserBase::Pos & pos, Expected & expected, bool id_mode, String & res)
     {
         return IParserBase::wrapParseImpl(pos, [&]
@@ -119,6 +105,17 @@ namespace
     }
 
 
+    bool parseSettingName(IParserBase::Pos & pos, Expected & expected, String & res)
+    {
+        ASTPtr name_ast;
+        if (!ParserCompoundIdentifier{}.parse(pos, name_ast, expected))
+            return false;
+
+        res = getIdentifierName(name_ast);
+        return true;
+    }
+
+
     bool parseSettingNameWithValueOrConstraints(
         IParserBase::Pos & pos,
         Expected & expected,
@@ -130,11 +127,10 @@ namespace
     {
         return IParserBase::wrapParseImpl(pos, [&]
         {
-            ASTPtr name_ast;
-            if (!ParserCompoundIdentifier{}.parse(pos, name_ast, expected))
+            String res_setting_name;
+            if (!parseSettingName(pos, expected, res_setting_name))
                 return false;
 
-            String res_setting_name = getIdentifierName(name_ast);
             std::optional<Field> res_value;
             std::optional<Field> res_min_value;
             std::optional<Field> res_max_value;
@@ -169,52 +165,112 @@ namespace
     }
 
 
-    bool parseSettingsProfileElement(IParserBase::Pos & pos,
-                                     Expected & expected,
-                                     bool id_mode,
-                                     bool use_inherit_keyword,
-                                     bool previous_element_was_parent_profile,
-                                     std::shared_ptr<ASTSettingsProfileElement> & result)
+    bool parseSettingsProfileElements(IParserBase::Pos & pos,
+                                      Expected & expected,
+                                      bool id_mode,
+                                      bool use_inherit_keyword,
+                                      std::vector<std::shared_ptr<ASTSettingsProfileElement>> & res)
     {
-        return IParserBase::wrapParseImpl(pos, [&]
+        std::vector<std::shared_ptr<ASTSettingsProfileElement>> elements;
+        bool found_none = false;
+
+        bool expect_profiles = false;
+        bool expect_settings = false;
+
+        auto parse_element = [&]
         {
-            String parent_profile;
-            String setting_name;
-            std::optional<Field> value;
-            std::optional<Field> min_value;
-            std::optional<Field> max_value;
-            std::optional<SettingConstraintWritability> writability;
+            if (ParserKeyword{Keyword::SETTINGS}.ignore(pos, expected) || ParserKeyword{Keyword::SETTING}.ignore(pos, expected))
+            {
+                expect_settings = true;
+            }
 
-            bool ok = parseSettingNameWithValueOrConstraints(pos, expected, setting_name, value, min_value, max_value, writability);
+            bool expect_settings_next = expect_settings;
 
-            if (!ok && (parseProfileKeyword(pos, expected, use_inherit_keyword) || previous_element_was_parent_profile))
-                ok = parseProfileNameOrID(pos, expected, id_mode, parent_profile);
+            if (ParserKeyword{Keyword::PROFILES}.ignore(pos, expected) || ParserKeyword{Keyword::PROFILE}.ignore(pos, expected))
+            {
+                expect_profiles = true;
+                expect_settings = false;
+            }
+            else if (use_inherit_keyword && ParserKeyword{Keyword::INHERIT}.ignore(pos, expected))
+            {
+                if (!ParserKeyword{Keyword::PROFILES}.ignore(pos, expected))
+                    ParserKeyword{Keyword::PROFILE}.ignore(pos, expected);
+                expect_profiles = true;
+                expect_settings = false;
+            }
 
-            if (!ok)
+            if (!expect_profiles && !expect_settings)
                 return false;
 
-            result = std::make_shared<ASTSettingsProfileElement>();
-            result->parent_profile = std::move(parent_profile);
-            result->setting_name = std::move(setting_name);
-            result->value = std::move(value);
-            result->min_value = std::move(min_value);
-            result->max_value = std::move(max_value);
-            result->writability = writability;
-            result->id_mode = id_mode;
-            result->use_inherit_keyword = use_inherit_keyword;
-            return true;
-        });
+            if (ParserKeyword{Keyword::NONE}.ignore(pos, expected))
+            {
+                found_none = true;
+                expect_settings = expect_settings_next;
+                return true;
+            }
+
+            if (expect_settings)
+            {
+                String setting_name;
+                std::optional<Field> value;
+                std::optional<Field> min_value;
+                std::optional<Field> max_value;
+                std::optional<SettingConstraintWritability> writability;
+                if (parseSettingNameWithValueOrConstraints(pos, expected, setting_name, value, min_value, max_value, writability))
+                {
+                    auto element = std::make_shared<ASTSettingsProfileElement>();
+                    element->setting_name = std::move(setting_name);
+                    element->value = std::move(value);
+                    element->min_value = std::move(min_value);
+                    element->max_value = std::move(max_value);
+                    element->writability = std::move(writability);
+                    elements.push_back(element);
+                    expect_profiles = false;
+                    expect_settings = expect_settings_next;
+                    return true;
+                }
+            }
+
+            if (expect_profiles)
+            {
+                String profile_name;
+                if (parseProfileNameOrID(pos, expected, id_mode, profile_name))
+                {
+                    auto element = std::make_shared<ASTSettingsProfileElement>();
+                    element->parent_profile = std::move(profile_name);
+                    element->id_mode = id_mode;
+                    element->use_inherit_keyword = use_inherit_keyword;
+                    elements.push_back(element);
+                    expect_settings = expect_settings_next;
+                    return true;
+                }
+            }
+
+            return false;
+        };
+
+        if (!ParserList::parseUtil(pos, expected, parse_element, false))
+            return false;
+
+        if (elements.empty() && !found_none)
+            return false;
+
+        res = std::move(elements);
+        return true;
     }
 }
 
 
 bool ParserSettingsProfileElement::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
 {
-    std::shared_ptr<ASTSettingsProfileElement> res;
-    if (!parseSettingsProfileElement(pos, expected, id_mode, use_inherit_keyword, false, res))
+    std::vector<std::shared_ptr<ASTSettingsProfileElement>> elements;
+    if (!parseSettingsProfileElements(pos, expected, id_mode, use_inherit_keyword, elements))
         return false;
 
-    node = res;
+    if (elements.size() != 1)
+        return false;
+
+    node = elements[0];
     return true;
 }
 
@@ -222,28 +278,8 @@ bool ParserSettingsProfileElement::parseImpl(Pos & pos, ASTPtr & node, Expected
 bool ParserSettingsProfileElements::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
 {
     std::vector<std::shared_ptr<ASTSettingsProfileElement>> elements;
-
-    if (ParserKeyword{Keyword::NONE}.ignore(pos, expected))
-    {
-    }
-    else
-    {
-        bool previous_element_was_parent_profile = false;
-
-        auto parse_element = [&]
-        {
-            std::shared_ptr<ASTSettingsProfileElement> element;
-            if (!parseSettingsProfileElement(pos, expected, id_mode, use_inherit_keyword, previous_element_was_parent_profile, element))
-                return false;
-
-            elements.push_back(element);
-            previous_element_was_parent_profile = !element->parent_profile.empty();
-            return true;
-        };
-
-        if (!ParserList::parseUtil(pos, expected, parse_element, false))
-            return false;
-    }
+    if (!parseSettingsProfileElements(pos, expected, id_mode, use_inherit_keyword, elements))
+        return false;
 
     auto result = std::make_shared<ASTSettingsProfileElements>();
     result->elements = std::move(elements);
@@ -251,4 +287,148 @@ bool ParserSettingsProfileElements::parseImpl(Pos & pos, ASTPtr & node, Expected
     return true;
 }
 
+
+bool ParserAlterSettingsProfileElements::parseImpl(Pos & pos, ASTPtr & node, Expected & expected)
+{
+    std::vector<std::shared_ptr<ASTSettingsProfileElement>> add_settings;
+    std::vector<std::shared_ptr<ASTSettingsProfileElement>> modify_settings;
+    std::vector<std::shared_ptr<ASTSettingsProfileElement>> drop_settings;
+    bool drop_all_settings = false;
+    bool drop_all_profiles = false;
+
+    std::vector<std::shared_ptr<ASTSettingsProfileElement>> old_style_settings;
+    if (parseSettingsProfileElements(pos, expected, /* id_mode= */ false, use_inherit_keyword, old_style_settings))
+    {
+        /// old style: "SETTINGS ..." replaces all the settings ad profiles.
+        add_settings = std::move(old_style_settings);
+        drop_all_settings = true;
+        drop_all_profiles = true;
+    }
+    else
+    {
+        /// new style: "MODIFY SETTINGS ..., ADD PROFILES ..., DROP PROFILES ..., DROP SETTINGS ..."
+        std::string_view action;
+        std::string_view target;
+
+        auto parse_element = [&]
+        {
+            if (ParserKeyword{Keyword::ADD}.ignore(pos, expected))
+            {
+                action = "ADD";
+                target = "";
+            }
+            else if (ParserKeyword{Keyword::DROP}.ignore(pos, expected))
+            {
+                action = "DROP";
+                target = "";
+            }
+            else if (ParserKeyword{Keyword::MODIFY}.ignore(pos, expected))
+            {
+                action = "MODIFY";
+                target = "";
+            }
+
+            if (!action.empty())
+            {
+                if (ParserKeyword{Keyword::ALL_PROFILES}.ignore(pos, expected))
+                    target = "ALL PROFILES";
+                else if (ParserKeyword{Keyword::ALL_SETTINGS}.ignore(pos, expected))
+                    target = "ALL SETTINGS";
+                else if (ParserKeyword{Keyword::PROFILES}.ignore(pos, expected) || ParserKeyword{Keyword::PROFILE}.ignore(pos, expected))
+                    target = "PROFILES";
+                else if (ParserKeyword{Keyword::SETTINGS}.ignore(pos, expected) || ParserKeyword{Keyword::SETTING}.ignore(pos, expected))
+                    target = "SETTINGS";
+            }
+
+            if (target.empty())
+                return false;
+
+            if (target == "PROFILES")
+            {
+                auto element = std::make_shared<ASTSettingsProfileElement>();
+                if (!parseProfileNameOrID(pos, expected, /* id_mode= */ false, element->parent_profile))
+                    return false;
+                if (action == "ADD")
+                {
+                    add_settings.push_back(element);
+                    return true;
+                }
+                if (action == "DROP")
+                {
+                    drop_settings.push_back(element);
+                    return true;
+                }
+                return false;
+            }
+
+            if (target == "SETTINGS")
+            {
+                auto element = std::make_shared<ASTSettingsProfileElement>();
+                if (action == "ADD" || action == "MODIFY")
+                {
+                    if (!parseSettingNameWithValueOrConstraints(pos, expected, element->setting_name, element->value, element->min_value, element->max_value, element->writability))
+                        return false;
+                    if (action == "ADD")
+                        add_settings.push_back(element);
+                    else
+                        modify_settings.push_back(element);
+                    return true;
+                }
+                if (action == "DROP")
+                {
+                    ASTPtr name_ast;
+                    if (!ParserCompoundIdentifier{}.parse(pos, name_ast, expected))
+                        return false;
+                    element->setting_name = getIdentifierName(name_ast);
+                    drop_settings.push_back(element);
+                    return true;
+                }
+                return false;
+            }
+
+            if (action == "DROP" && target == "ALL PROFILES")
+            {
+                drop_all_profiles = true;
+                return true;
+            }
+
+            if (action == "DROP" && target == "ALL SETTINGS")
+            {
+                drop_all_settings = true;
+                return true;
+            }
+
+            return false;
+        };
+
+        if (!ParserList::parseUtil(pos, expected, parse_element, false))
+            return false;
+    }
+
+    if (add_settings.empty() && modify_settings.empty() && drop_settings.empty() && !drop_all_settings && !drop_all_profiles)
+        return false;
+
+    auto result = std::make_shared<ASTAlterSettingsProfileElements>();
+    if (!add_settings.empty())
+    {
+        result->add_settings = std::make_shared<ASTSettingsProfileElements>();
+        result->add_settings->elements = std::move(add_settings);
+    }
+    if (!modify_settings.empty())
+    {
+        result->modify_settings = std::make_shared<ASTSettingsProfileElements>();
+        result->modify_settings->elements = std::move(modify_settings);
+    }
+    if (!drop_settings.empty())
+    {
+        result->drop_settings = std::make_shared<ASTSettingsProfileElements>();
+        result->drop_settings->elements = std::move(drop_settings);
+    }
+    result->drop_all_settings = drop_all_settings;
+    result->drop_all_profiles = drop_all_profiles;
+
+    node = result;
+    return true;
+}
+
 }

src/Parsers/Access/ParserSettingsProfileElement.h

@@ -39,4 +39,18 @@ private:
     bool use_inherit_keyword = false;
 };
 
+
+class ParserAlterSettingsProfileElements : public IParserBase
+{
+public:
+    ParserAlterSettingsProfileElements & useInheritKeyword(bool use_inherit_keyword_ = true) { use_inherit_keyword = use_inherit_keyword_; return *this; }
+
+protected:
+    const char * getName() const override { return "AlterSettingsProfileElements"; }
+    bool parseImpl(Pos & pos, ASTPtr & node, Expected & expected) override;
+
+private:
+    bool use_inherit_keyword = false;
+};
+
 }

src/Parsers/CommonParsers.h

@@ -10,22 +10,25 @@ namespace DB
 {
 
 #define APPLY_FOR_PARSER_KEYWORDS(MR_MACROS) \
+    MR_MACROS(ADD, "ADD") \
     MR_MACROS(ADD_COLUMN, "ADD COLUMN") \
     MR_MACROS(ADD_CONSTRAINT, "ADD CONSTRAINT") \
     MR_MACROS(ADD_INDEX, "ADD INDEX") \
     MR_MACROS(ADD_PROJECTION, "ADD PROJECTION") \
     MR_MACROS(ADD_STATISTICS, "ADD STATISTICS") \
-    MR_MACROS(ADD, "ADD") \
     MR_MACROS(ADMIN_OPTION_FOR, "ADMIN OPTION FOR") \
     MR_MACROS(AFTER, "AFTER") \
     MR_MACROS(ALGORITHM, "ALGORITHM") \
     MR_MACROS(ALIAS, "ALIAS") \
     MR_MACROS(ALL, "ALL") \
+    MR_MACROS(ALL_PROFILES, "ALL PROFILES") \
+    MR_MACROS(ALL_SETTINGS, "ALL SETTINGS") \
     MR_MACROS(ALTER_COLUMN, "ALTER COLUMN") \
     MR_MACROS(ALTER_DATABASE, "ALTER DATABASE") \
     MR_MACROS(ALTER_LIVE_VIEW, "ALTER LIVE VIEW") \
     MR_MACROS(ALTER_POLICY, "ALTER POLICY") \
     MR_MACROS(ALTER_PROFILE, "ALTER PROFILE") \
+    MR_MACROS(ALTER_PROFILES, "ALTER PROFILES") \
     MR_MACROS(ALTER_QUOTA, "ALTER QUOTA") \
     MR_MACROS(ALTER_ROLE, "ALTER ROLE") \
     MR_MACROS(ALTER_ROW_POLICY, "ALTER ROW POLICY") \
@@ -380,6 +383,7 @@ namespace DB
     MR_MACROS(PRIMARY_KEY, "PRIMARY KEY") \
     MR_MACROS(PRIMARY, "PRIMARY") \
     MR_MACROS(PROFILE, "PROFILE") \
+    MR_MACROS(PROFILES, "PROFILES") \
     MR_MACROS(PROJECTION, "PROJECTION") \
     MR_MACROS(PROTOBUF, "Protobuf") \
     MR_MACROS(PULL, "PULL") \
@@ -444,7 +448,8 @@ namespace DB
     MR_MACROS(SET_ROLE_DEFAULT, "SET ROLE DEFAULT") \
     MR_MACROS(SET_ROLE, "SET ROLE") \
     MR_MACROS(SET_TRANSACTION_SNAPSHOT, "SET TRANSACTION SNAPSHOT") \
-    MR_MACROS(SET, "SET") \
+    MR_MACROS(SET, "SET")                    \
+    MR_MACROS(SETTING, "SETTING") \
     MR_MACROS(SETTINGS, "SETTINGS") \
     MR_MACROS(SHOW_ACCESS, "SHOW ACCESS") \
     MR_MACROS(SHOW_CREATE, "SHOW CREATE") \

src/TableFunctions/TableFunctionObjectStorage.cpp

@@ -226,6 +226,26 @@ template class TableFunctionObjectStorage<HDFSClusterDefinition, StorageHDFSConf
 #endif
 template class TableFunctionObjectStorage<LocalDefinition, StorageLocalConfiguration>;
 
+#if USE_AVRO && USE_AWS_S3
+template class TableFunctionObjectStorage<IcebergS3ClusterDefinition, StorageS3IcebergConfiguration>;
+#endif
+
+#if USE_AVRO && USE_AZURE_BLOB_STORAGE
+template class TableFunctionObjectStorage<IcebergAzureClusterDefinition, StorageAzureIcebergConfiguration>;
+#endif
+
+#if USE_AVRO && USE_HDFS
+template class TableFunctionObjectStorage<IcebergHDFSClusterDefinition, StorageHDFSIcebergConfiguration>;
+#endif
+
+#if USE_PARQUET && USE_AWS_S3
+template class TableFunctionObjectStorage<DeltaLakeClusterDefinition, StorageS3DeltaLakeConfiguration>;
+#endif
+
+#if USE_AWS_S3
+template class TableFunctionObjectStorage<HudiClusterDefinition, StorageS3HudiConfiguration>;
+#endif
+
 #if USE_AVRO
 void registerTableFunctionIceberg(TableFunctionFactory & factory)
 {

src/TableFunctions/TableFunctionObjectStorageCluster.cpp

@@ -96,7 +96,7 @@ void registerTableFunctionObjectStorageCluster(TableFunctionFactory & factory)
     {
         .documentation = {
             .description=R"(The table function can be used to read the data stored on HDFS in parallel for many nodes in a specified cluster.)",
-            .examples{{"HDFSCluster", "SELECT * FROM HDFSCluster(cluster_name, uri, format)", ""}}},
+            .examples{{"HDFSCluster", "SELECT * FROM HDFSCluster(cluster, uri, format)", ""}}},
             .allow_readonly = false
         }
     );
@@ -105,15 +105,77 @@ void registerTableFunctionObjectStorageCluster(TableFunctionFactory & factory)
     UNUSED(factory);
 }
 
+
+#if USE_AVRO
+void registerTableFunctionIcebergCluster(TableFunctionFactory & factory)
+{
+    UNUSED(factory);
+
 #if USE_AWS_S3
-template class TableFunctionObjectStorageCluster<S3ClusterDefinition, StorageS3Configuration>;
+    factory.registerFunction<TableFunctionIcebergS3Cluster>(
+        {.documentation
+         = {.description = R"(The table function can be used to read the Iceberg table stored on S3 object store in parallel for many nodes in a specified cluster.)",
+            .examples{{"icebergS3Cluster", "SELECT * FROM icebergS3Cluster(cluster, url, [, NOSIGN | access_key_id, secret_access_key, [session_token]], format, [,compression])", ""}},
+            .categories{"DataLake"}},
+         .allow_readonly = false});
 #endif
 
 #if USE_AZURE_BLOB_STORAGE
-template class TableFunctionObjectStorageCluster<AzureClusterDefinition, StorageAzureConfiguration>;
+    factory.registerFunction<TableFunctionIcebergAzureCluster>(
+        {.documentation
+         = {.description = R"(The table function can be used to read the Iceberg table stored on Azure object store in parallel for many nodes in a specified cluster.)",
+            .examples{{"icebergAzureCluster", "SELECT * FROM icebergAzureCluster(cluster, connection_string|storage_account_url, container_name, blobpath, [account_name, account_key, format, compression])", ""}},
+            .categories{"DataLake"}},
+         .allow_readonly = false});
 #endif
 
 #if USE_HDFS
-template class TableFunctionObjectStorageCluster<HDFSClusterDefinition, StorageHDFSConfiguration>;
+    factory.registerFunction<TableFunctionIcebergHDFSCluster>(
+        {.documentation
+         = {.description = R"(The table function can be used to read the Iceberg table stored on HDFS virtual filesystem in parallel for many nodes in a specified cluster.)",
+            .examples{{"icebergHDFSCluster", "SELECT * FROM icebergHDFSCluster(cluster, uri, [format], [structure], [compression_method])", ""}},
+            .categories{"DataLake"}},
+         .allow_readonly = false});
 #endif
 }
+#endif
+
+#if USE_AWS_S3
+#if USE_PARQUET
+void registerTableFunctionDeltaLakeCluster(TableFunctionFactory & factory)
+{
+    factory.registerFunction<TableFunctionDeltaLakeCluster>(
+        {.documentation
+         = {.description = R"(The table function can be used to read the DeltaLake table stored on object store in parallel for many nodes in a specified cluster.)",
+            .examples{{"deltaLakeCluster", "SELECT * FROM deltaLakeCluster(cluster, url, access_key_id, secret_access_key)", ""}},
+            .categories{"DataLake"}},
+         .allow_readonly = false});
+}
+#endif
+
+void registerTableFunctionHudiCluster(TableFunctionFactory & factory)
+{
+    factory.registerFunction<TableFunctionHudiCluster>(
+        {.documentation
+         = {.description = R"(The table function can be used to read the Hudi table stored on object store in parallel for many nodes in a specified cluster.)",
+            .examples{{"hudiCluster", "SELECT * FROM hudiCluster(cluster, url, access_key_id, secret_access_key)", ""}},
+            .categories{"DataLake"}},
+         .allow_readonly = false});
+}
+#endif
+
+void registerDataLakeClusterTableFunctions(TableFunctionFactory & factory)
+{
+    UNUSED(factory);
+#if USE_AVRO
+    registerTableFunctionIcebergCluster(factory);
+#endif
+#if USE_AWS_S3
+#if USE_PARQUET
+    registerTableFunctionDeltaLakeCluster(factory);
+#endif
+    registerTableFunctionHudiCluster(factory);
+#endif
+}
+
+}

src/TableFunctions/TableFunctionObjectStorageCluster.h

@@ -33,6 +33,36 @@ struct HDFSClusterDefinition
     static constexpr auto storage_type_name = "HDFSCluster";
 };
 
+struct IcebergS3ClusterDefinition
+{
+    static constexpr auto name = "icebergS3Cluster";
+    static constexpr auto storage_type_name = "IcebergS3Cluster";
+};
+
+struct IcebergAzureClusterDefinition
+{
+    static constexpr auto name = "icebergAzureCluster";
+    static constexpr auto storage_type_name = "IcebergAzureCluster";
+};
+
+struct IcebergHDFSClusterDefinition
+{
+    static constexpr auto name = "icebergHDFSCluster";
+    static constexpr auto storage_type_name = "IcebergHDFSCluster";
+};
+
+struct DeltaLakeClusterDefinition
+{
+    static constexpr auto name = "deltaLakeCluster";
+    static constexpr auto storage_type_name = "DeltaLakeS3Cluster";
+};
+
+struct HudiClusterDefinition
+{
+    static constexpr auto name = "hudiCluster";
+    static constexpr auto storage_type_name = "HudiS3Cluster";
+};
+
 /**
 * Class implementing s3/hdfs/azureBlobStorageCluster(...) table functions,
 * which allow to process many files from S3/HDFS/Azure blob storage on a specific cluster.
@@ -79,4 +109,25 @@ using TableFunctionAzureBlobCluster = TableFunctionObjectStorageCluster<AzureClu
 #if USE_HDFS
 using TableFunctionHDFSCluster = TableFunctionObjectStorageCluster<HDFSClusterDefinition, StorageHDFSConfiguration>;
 #endif
+
+#if USE_AVRO && USE_AWS_S3
+using TableFunctionIcebergS3Cluster = TableFunctionObjectStorageCluster<IcebergS3ClusterDefinition, StorageS3IcebergConfiguration>;
+#endif
+
+#if USE_AVRO && USE_AZURE_BLOB_STORAGE
+using TableFunctionIcebergAzureCluster = TableFunctionObjectStorageCluster<IcebergAzureClusterDefinition, StorageAzureIcebergConfiguration>;
+#endif
+
+#if USE_AVRO && USE_HDFS
+using TableFunctionIcebergHDFSCluster = TableFunctionObjectStorageCluster<IcebergHDFSClusterDefinition, StorageHDFSIcebergConfiguration>;
+#endif
+
+#if USE_AWS_S3 && USE_PARQUET
+using TableFunctionDeltaLakeCluster = TableFunctionObjectStorageCluster<DeltaLakeClusterDefinition, StorageS3DeltaLakeConfiguration>;
+#endif
+
+#if USE_AWS_S3
+using TableFunctionHudiCluster = TableFunctionObjectStorageCluster<HudiClusterDefinition, StorageS3HudiConfiguration>;
+#endif
+
 }

src/TableFunctions/registerTableFunctions.cpp

@@ -66,6 +66,7 @@ void registerTableFunctions(bool use_legacy_mongodb_integration [[maybe_unused]]
     registerTableFunctionObjectStorage(factory);
     registerTableFunctionObjectStorageCluster(factory);
     registerDataLakeTableFunctions(factory);
+    registerDataLakeClusterTableFunctions(factory);
 }
 
 }

src/TableFunctions/registerTableFunctions.h

@@ -70,6 +70,7 @@ void registerTableFunctionExplain(TableFunctionFactory & factory);
 void registerTableFunctionObjectStorage(TableFunctionFactory & factory);
 void registerTableFunctionObjectStorageCluster(TableFunctionFactory & factory);
 void registerDataLakeTableFunctions(TableFunctionFactory & factory);
+void registerDataLakeClusterTableFunctions(TableFunctionFactory & factory);
 
 void registerTableFunctionTimeSeries(TableFunctionFactory & factory);
 

tests/integration/test_storage_iceberg/configs/config.d/cluster.xml

@@ -0,0 +1,20 @@
+<clickhouse>
+    <remote_servers>
+        <cluster_simple>
+            <shard>
+                <replica>
+                    <host>node1</host>
+                    <port>9000</port>
+                </replica>
+                <replica>
+                    <host>node2</host>
+                    <port>9000</port>
+                </replica>
+                <replica>
+                    <host>node3</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </cluster_simple>
+    </remote_servers>
+</clickhouse>

tests/integration/test_storage_iceberg/configs/config.d/query_log.xml

@@ -0,0 +1,6 @@
+<clickhouse>
+    <query_log>
+      <database>system</database>
+      <table>query_log</table>
+    </query_log>
+</clickhouse>

tests/integration/test_storage_iceberg/test.py

@@ -73,14 +73,38 @@ def started_cluster():
         cluster.add_instance(
             "node1",
             main_configs=[
+                "configs/config.d/query_log.xml",
+                "configs/config.d/cluster.xml",
                 "configs/config.d/named_collections.xml",
                 "configs/config.d/filesystem_caches.xml",
             ],
             user_configs=["configs/users.d/users.xml"],
             with_minio=True,
             with_azurite=True,
-            stay_alive=True,
             with_hdfs=with_hdfs,
+            stay_alive=True,
+        )
+        cluster.add_instance(
+            "node2",
+            main_configs=[
+                "configs/config.d/query_log.xml",
+                "configs/config.d/cluster.xml",
+                "configs/config.d/named_collections.xml",
+                "configs/config.d/filesystem_caches.xml",
+            ],
+            user_configs=["configs/users.d/users.xml"],
+            stay_alive=True,
+        )
+        cluster.add_instance(
+            "node3",
+            main_configs=[
+                "configs/config.d/query_log.xml",
+                "configs/config.d/cluster.xml",
+                "configs/config.d/named_collections.xml",
+                "configs/config.d/filesystem_caches.xml",
+            ],
+            user_configs=["configs/users.d/users.xml"],
+            stay_alive=True,
         )
 
         logging.info("Starting cluster...")
@@ -182,6 +206,7 @@ def get_creation_expression(
     cluster,
     format="Parquet",
     table_function=False,
+    run_on_cluster=False,
     **kwargs,
 ):
     if storage_type == "s3":
@@ -189,35 +214,56 @@ def get_creation_expression(
             bucket = kwargs["bucket"]
         else:
             bucket = cluster.minio_bucket
-        print(bucket)
-        if table_function:
-            return f"icebergS3(s3, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"
+
+        if run_on_cluster:
+            assert table_function
+            return f"icebergS3Cluster('cluster_simple', s3, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"
         else:
-            return f"""
-                DROP TABLE IF EXISTS {table_name};
-                CREATE TABLE {table_name}
-                ENGINE=IcebergS3(s3, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"""
+            if table_function:
+                return f"icebergS3(s3, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"
+            else:
+                return f"""
+                    DROP TABLE IF EXISTS {table_name};
+                    CREATE TABLE {table_name}
+                    ENGINE=IcebergS3(s3, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'http://minio1:9001/{bucket}/')"""
+
     elif storage_type == "azure":
-        if table_function:
+        if run_on_cluster:
+            assert table_function
             return f"""
-                icebergAzure(azure, container = '{cluster.azure_container_name}', storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})
+                icebergAzureCluster('cluster_simple', azure, container = '{cluster.azure_container_name}', storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})
             """
         else:
-            return f"""
-                DROP TABLE IF EXISTS {table_name};
-                CREATE TABLE {table_name}
-                ENGINE=IcebergAzure(azure, container = {cluster.azure_container_name}, storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})"""
+            if table_function:
+                return f"""
+                    icebergAzure(azure, container = '{cluster.azure_container_name}', storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})
+                """
+            else:
+                return f"""
+                    DROP TABLE IF EXISTS {table_name};
+                    CREATE TABLE {table_name}
+                    ENGINE=IcebergAzure(azure, container = {cluster.azure_container_name}, storage_account_url = '{cluster.env_variables["AZURITE_STORAGE_ACCOUNT_URL"]}', blob_path = '/iceberg_data/default/{table_name}/', format={format})"""
+
     elif storage_type == "hdfs":
-        if table_function:
+        if run_on_cluster:
+            assert table_function
             return f"""
-                icebergHDFS(hdfs, filename= 'iceberg_data/default/{table_name}/', format={format}, url = 'hdfs://hdfs1:9000/')
+                icebergHDFSCluster('cluster_simple', hdfs, filename= 'iceberg_data/default/{table_name}/', format={format}, url = 'hdfs://hdfs1:9000/')
             """
         else:
-            return f"""
-                DROP TABLE IF EXISTS {table_name};
-                CREATE TABLE {table_name}
-                ENGINE=IcebergHDFS(hdfs, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'hdfs://hdfs1:9000/');"""
+            if table_function:
+                return f"""
+                    icebergHDFS(hdfs, filename= 'iceberg_data/default/{table_name}/', format={format}, url = 'hdfs://hdfs1:9000/')
+                """
+            else:
+                return f"""
+                    DROP TABLE IF EXISTS {table_name};
+                    CREATE TABLE {table_name}
+                    ENGINE=IcebergHDFS(hdfs, filename = 'iceberg_data/default/{table_name}/', format={format}, url = 'hdfs://hdfs1:9000/');"""
+
     elif storage_type == "local":
+        assert not run_on_cluster
+
         if table_function:
             return f"""
                 icebergLocal(local, path = '/iceberg_data/default/{table_name}/', format={format})
@@ -227,6 +273,7 @@ def get_creation_expression(
                 DROP TABLE IF EXISTS {table_name};
                 CREATE TABLE {table_name}
                 ENGINE=IcebergLocal(local, path = '/iceberg_data/default/{table_name}/', format={format});"""
+
     else:
         raise Exception(f"Unknown iceberg storage type: {storage_type}")
 
@@ -492,6 +539,108 @@ def test_types(started_cluster, format_version, storage_type):
     )
 
 
+@pytest.mark.parametrize("format_version", ["1", "2"])
+@pytest.mark.parametrize("storage_type", ["s3", "azure", "hdfs"])
+def test_cluster_table_function(started_cluster, format_version, storage_type):
+    if is_arm() and storage_type == "hdfs":
+        pytest.skip("Disabled test IcebergHDFS for aarch64")
+
+    instance = started_cluster.instances["node1"]
+    spark = started_cluster.spark_session
+
+    TABLE_NAME = (
+        "test_iceberg_cluster_"
+        + format_version
+        + "_"
+        + storage_type
+        + "_"
+        + get_uuid_str()
+    )
+
+    def add_df(mode):
+        write_iceberg_from_df(
+            spark,
+            generate_data(spark, 0, 100),
+            TABLE_NAME,
+            mode=mode,
+            format_version=format_version,
+        )
+
+        files = default_upload_directory(
+            started_cluster,
+            storage_type,
+            f"/iceberg_data/default/{TABLE_NAME}/",
+            f"/iceberg_data/default/{TABLE_NAME}/",
+        )
+
+        logging.info(f"Adding another dataframe. result files: {files}")
+
+        return files
+
+    files = add_df(mode="overwrite")
+    for i in range(1, len(started_cluster.instances)):
+        files = add_df(mode="append")
+
+    logging.info(f"Setup complete. files: {files}")
+    assert len(files) == 5 + 4 * (len(started_cluster.instances) - 1)
+
+    clusters = instance.query(f"SELECT * FROM system.clusters")
+    logging.info(f"Clusters setup: {clusters}")
+
+    # Regular Query only node1
+    table_function_expr = get_creation_expression(
+        storage_type, TABLE_NAME, started_cluster, table_function=True
+    )
+    select_regular = (
+        instance.query(f"SELECT * FROM {table_function_expr}").strip().split()
+    )
+
+    # Cluster Query with node1 as coordinator
+    table_function_expr_cluster = get_creation_expression(
+        storage_type,
+        TABLE_NAME,
+        started_cluster,
+        table_function=True,
+        run_on_cluster=True,
+    )
+    select_cluster = (
+        instance.query(f"SELECT * FROM {table_function_expr_cluster}").strip().split()
+    )
+
+    # Simple size check
+    assert len(select_regular) == 600
+    assert len(select_cluster) == 600
+
+    # Actual check
+    assert select_cluster == select_regular
+
+    # Check query_log
+    for replica in started_cluster.instances.values():
+        replica.query("SYSTEM FLUSH LOGS")
+
+    for node_name, replica in started_cluster.instances.items():
+        cluster_secondary_queries = (
+            replica.query(
+                f"""
+                SELECT query, type, is_initial_query, read_rows, read_bytes FROM system.query_log
+                WHERE
+                    type = 'QueryStart' AND
+                    positionCaseInsensitive(query, '{storage_type}Cluster') != 0 AND
+                    position(query, '{TABLE_NAME}') != 0 AND
+                    position(query, 'system.query_log') = 0 AND
+                    NOT is_initial_query
+            """
+            )
+            .strip()
+            .split("\n")
+        )
+
+        logging.info(
+            f"[{node_name}] cluster_secondary_queries: {cluster_secondary_queries}"
+        )
+        assert len(cluster_secondary_queries) == 1
+
+
 @pytest.mark.parametrize("format_version", ["1", "2"])
 @pytest.mark.parametrize("storage_type", ["s3", "azure", "hdfs", "local"])
 def test_delete_files(started_cluster, format_version, storage_type):

tests/queries/0_stateless/01294_create_settings_profile.reference

@@ -49,7 +49,7 @@ CREATE SETTINGS PROFILE `s4_01294` TO r1_01294
 CREATE SETTINGS PROFILE `s1_01294` SETTINGS readonly = 1
 CREATE SETTINGS PROFILE `s2_01294` SETTINGS readonly CONST
 CREATE SETTINGS PROFILE `s3_01294` SETTINGS INHERIT `readonly`
-CREATE SETTINGS PROFILE `s4_01294` SETTINGS INHERIT `readonly`, INHERIT `readonly`
+CREATE SETTINGS PROFILE `s4_01294` SETTINGS INHERIT `readonly`
 CREATE SETTINGS PROFILE `s5_01294` SETTINGS INHERIT `readonly`, readonly = 1
 CREATE SETTINGS PROFILE `s6_01294` SETTINGS INHERIT `readonly`, readonly CONST
 -- system.settings_profiles

tests/queries/0_stateless/02943_alter_user_modify_profiles.reference

@@ -0,0 +1,21 @@
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, PROFILE `profile_b`
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_b`, PROFILE `profile_a`, PROFILE `profile_c`
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, PROFILE `profile_d`
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_e`
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, PROFILE `profile_b`
+CREATE ROLE test_role SETTINGS PROFILE `profile_b`, PROFILE `profile_a`, PROFILE `profile_c`
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, PROFILE `profile_d`
+CREATE ROLE test_role SETTINGS PROFILE `profile_e`
+CREATE ROLE test_role
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, INHERIT `profile_b`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_b`, INHERIT `profile_a`, INHERIT `profile_c`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, INHERIT `profile_d`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_e`
+CREATE SETTINGS PROFILE `test_profile`

tests/queries/0_stateless/02943_alter_user_modify_profiles.sh

@@ -0,0 +1,76 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+test_user="test_user_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+test_role="test_role_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+test_profile="test_profile_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+
+profile_a="profile_a_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+profile_b="profile_b_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+profile_c="profile_c_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+profile_d="profile_d_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+profile_e="profile_e_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_a};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_b};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_c};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_d};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_e};"
+
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_a};"
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_b};"
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_c};"
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_d};"
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_e};"
+
+function show_create()
+{
+    local type="$1"
+    local name="$2"
+
+    $CLICKHOUSE_CLIENT -q "SHOW CREATE $type ${name};" | sed -e "s/${name}/test_${type}/g" -e "s/${profile_a}/profile_a/g" -e "s/${profile_b}/profile_b/g" -e "s/${profile_c}/profile_c/g" -e "s/${profile_d}/profile_d/g" -e "s/${profile_e}/profile_e/g"
+}
+
+function run_test()
+{
+    local type="$1"
+    local name="$2"
+
+    $CLICKHOUSE_CLIENT -q "DROP $type IF EXISTS ${name};"
+
+    $CLICKHOUSE_CLIENT -q "CREATE $type ${name};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD PROFILE ${profile_a};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD PROFILE ${profile_b};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD PROFILES ${profile_a}, ${profile_c};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD PROFILE ${profile_d} DROP PROFILES ${profile_b}, ${profile_c};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL PROFILES, ADD PROFILE ${profile_e};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL PROFILES;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "DROP $type ${name};"
+}
+
+run_test user ${test_user}
+run_test role ${test_role}
+run_test profile ${test_profile}
+
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_a};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_b};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_c};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_d};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_e};"

tests/queries/0_stateless/02943_alter_user_modify_profiles_and_settings.reference

@@ -0,0 +1,39 @@
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_x = 123, custom_t = 7, custom_s = \'str\'
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_a`, custom_x = 123, custom_t = 7, custom_s = \'str\'
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS PROFILE `profile_b`, PROFILE `profile_a`, custom_x = 321, custom_s = \'str\' CONST
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, PROFILE `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS custom_x = 123, custom_t = 7, custom_s = \'str\'
+CREATE ROLE test_role SETTINGS PROFILE `profile_a`, custom_x = 123, custom_t = 7, custom_s = \'str\'
+CREATE ROLE test_role SETTINGS PROFILE `profile_b`, PROFILE `profile_a`, custom_x = 321, custom_s = \'str\' CONST
+CREATE ROLE test_role
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, INHERIT `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, INHERIT `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, INHERIT `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, INHERIT `profile_b`, custom_x = 123, custom_y = 56.5, custom_w = \'www\'
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_x = 123, custom_t = 7, custom_s = \'str\'
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_a`, custom_x = 123, custom_t = 7, custom_s = \'str\'
+CREATE SETTINGS PROFILE `test_profile` SETTINGS INHERIT `profile_b`, INHERIT `profile_a`, custom_x = 321, custom_s = \'str\' CONST
+CREATE SETTINGS PROFILE `test_profile`

tests/queries/0_stateless/02943_alter_user_modify_profiles_and_settings.sh

@@ -0,0 +1,86 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+test_user="test_user_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+test_role="test_role_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+test_profile="test_profile_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+
+profile_a="profile_a_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+profile_b="profile_b_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+profile_c="profile_c_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_a};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_b};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE IF EXISTS ${profile_c};"
+
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_a};"
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_b};"
+$CLICKHOUSE_CLIENT -q "CREATE PROFILE ${profile_c};"
+
+function show_create()
+{
+    local type="$1"
+    local name="$2"
+
+    $CLICKHOUSE_CLIENT -q "SHOW CREATE $type ${name};" | sed -e "s/${name}/test_${type}/g" -e "s/${profile_a}/profile_a/g" -e "s/${profile_b}/profile_b/g" -e "s/${profile_c}/profile_c/g"
+}
+
+function run_test()
+{
+    local type="$1"
+    local name="$2"
+
+    $CLICKHOUSE_CLIENT -q "DROP $type IF EXISTS ${name};"
+
+    $CLICKHOUSE_CLIENT -q "CREATE $type ${name};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTINGS custom_x = 123, custom_y=56.5, custom_w='www', ADD PROFILES ${profile_a}, ${profile_b}"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL SETTINGS, DROP ALL PROFILES"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTINGS custom_x = 123, custom_y=56.5, custom_w='www', PROFILES ${profile_a}, ${profile_b}"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL SETTINGS, ALL PROFILES"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_x = 123, ADD SETTING custom_y=56.5, ADD SETTING custom_w='www', ADD PROFILE ${profile_a}, ADD PROFILE ${profile_b}"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL PROFILES, DROP ALL SETTINGS"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_x = 123 ADD SETTING custom_y=56.5 ADD SETTING custom_w='www' ADD PROFILE ${profile_a} ADD PROFILE ${profile_b}"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL SETTINGS DROP ALL PROFILES"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_x = 123, custom_t = 7, custom_s = 'str'"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD PROFILE ${profile_a}"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD PROFILE ${profile_b}, ${profile_a}, DROP SETTING custom_t, MODIFY SETTING custom_s READONLY, custom_x = 321"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL SETTINGS, ALL PROFILES"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "DROP $type ${name};"
+}
+
+run_test user ${test_user}
+run_test role ${test_role}
+run_test profile ${test_profile}
+
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_a};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_b};"
+$CLICKHOUSE_CLIENT -q "DROP PROFILE ${profile_c};"

tests/queries/0_stateless/02943_alter_user_modify_settings.reference

@@ -0,0 +1,42 @@
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_a = 100
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_a = 100, custom_b = 200
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_a = 100, custom_b = 300, custom_c = 400
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_b = 300, custom_c = 400, custom_d = 500
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 0 MAX 1000
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 0 MAX 2000
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 500 MAX 2000
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 700 MIN 500 MAX 2000
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 800 MIN 150
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_c = 400, custom_d = 500
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_c = 400, custom_d = 500, custom_x = 1, custom_y = 2
+CREATE USER test_user IDENTIFIED WITH no_password SETTINGS custom_z = 3
+CREATE USER test_user IDENTIFIED WITH no_password
+CREATE ROLE test_role
+CREATE ROLE test_role SETTINGS custom_a = 100
+CREATE ROLE test_role SETTINGS custom_a = 100, custom_b = 200
+CREATE ROLE test_role SETTINGS custom_a = 100, custom_b = 300, custom_c = 400
+CREATE ROLE test_role SETTINGS custom_b = 300, custom_c = 400, custom_d = 500
+CREATE ROLE test_role SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 0 MAX 1000
+CREATE ROLE test_role SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 0 MAX 2000
+CREATE ROLE test_role SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 500 MAX 2000
+CREATE ROLE test_role SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 700 MIN 500 MAX 2000
+CREATE ROLE test_role SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 800 MIN 150
+CREATE ROLE test_role SETTINGS custom_c = 400, custom_d = 500
+CREATE ROLE test_role SETTINGS custom_c = 400, custom_d = 500, custom_x = 1, custom_y = 2
+CREATE ROLE test_role SETTINGS custom_z = 3
+CREATE ROLE test_role
+CREATE SETTINGS PROFILE `test_profile`
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_a = 100
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_a = 100, custom_b = 200
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_a = 100, custom_b = 300, custom_c = 400
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_b = 300, custom_c = 400, custom_d = 500
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 0 MAX 1000
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 0 MAX 2000
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 600 MIN 500 MAX 2000
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 700 MIN 500 MAX 2000
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_b = 300, custom_c = 400, custom_d = 500, custom_e = 800 MIN 150
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_c = 400, custom_d = 500
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_c = 400, custom_d = 500, custom_x = 1, custom_y = 2
+CREATE SETTINGS PROFILE `test_profile` SETTINGS custom_z = 3
+CREATE SETTINGS PROFILE `test_profile`

tests/queries/0_stateless/02943_alter_user_modify_settings.sh

@@ -0,0 +1,73 @@
+#!/usr/bin/env bash
+
+CURDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+# shellcheck source=../shell_config.sh
+. "$CURDIR"/../shell_config.sh
+
+test_user="test_user_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+test_role="test_role_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+test_profile="test_profile_02932_${CLICKHOUSE_DATABASE}_$RANDOM"
+
+function show_create()
+{
+    local type="$1"
+    local name="$2"
+
+    $CLICKHOUSE_CLIENT -q "SHOW CREATE $type ${name};" | sed "s/${name}/test_${type}/g"
+}
+
+function run_test()
+{
+    local type="$1"
+    local name="$2"
+
+    $CLICKHOUSE_CLIENT -q "DROP $type IF EXISTS ${name};"
+
+    $CLICKHOUSE_CLIENT -q "CREATE $type ${name};"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_a=100;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_b=200;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTINGS custom_b=300, custom_c=400;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_d=500 DROP SETTING custom_a;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_e=600 MIN 0 MAX 1000;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} MODIFY SETTING custom_e MAX 2000;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} MODIFY SETTING custom_e MIN 500;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} MODIFY SETTING custom_e=700;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} ADD SETTING custom_e=800 MIN 150;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP SETTINGS custom_b, custom_e;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} MODIFY SETTINGS custom_x=1, custom_y=2;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL SETTINGS, ADD SETTING custom_z=3;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "ALTER $type ${name} DROP ALL SETTINGS;"
+    show_create $type ${name}
+
+    $CLICKHOUSE_CLIENT -q "DROP $type ${name};"
+}
+
+run_test user ${test_user}
+run_test role ${test_role}
+run_test profile ${test_profile}

tests/queries/0_stateless/03216_arrayWithConstant_limits.sql

@@ -1,3 +1,6 @@
 SELECT arrayWithConstant(96142475, ['qMUF']); -- { serverError TOO_LARGE_ARRAY_SIZE }
 SELECT arrayWithConstant(100000000, materialize([[[[[[[[[['Hello, world!']]]]]]]]]])); -- { serverError TOO_LARGE_ARRAY_SIZE }
 SELECT length(arrayWithConstant(10000000, materialize([[[[[[[[[['Hello world']]]]]]]]]])));
+
+CREATE TEMPORARY TABLE args (value Array(Int)) ENGINE=Memory AS SELECT [1, 1, 1, 1] as value FROM numbers(1, 100);
+SELECT length(arrayWithConstant(1000000, value)) FROM args FORMAT NULL;

utils/check-style/aspell-ignore/en/aspell-dict.txt

@@ -244,7 +244,10 @@ Deduplication
 DefaultTableEngine
 DelayedInserts
 DeliveryTag
+Deltalake
 DeltaLake
+deltalakeCluster
+deltaLakeCluster
 Denormalize
 DestroyAggregatesThreads
 DestroyAggregatesThreadsActive
@@ -377,10 +380,15 @@ Homebrew's
 HorizontalDivide
 Hostname
 HouseOps
+hudi
 Hudi
+hudiCluster
+HudiCluster
 HyperLogLog
 Hypot
 IANA
+icebergCluster
+IcebergCluster
 IDE
 IDEs
 IDNA