Merge branch 'releases/25.8.12' into backports/25.8.12/89367

Author: zvonand

ci/jobs/scripts/check_style/aspell-ignore/en/aspell-dict.txt

@@ -1422,6 +1422,7 @@ atanh
 atomicity
 auth
 authType
+authenticatedUser
 authenticator
 authenticators
 autocompletion

ci/jobs/scripts/check_style/various_checks.sh

@@ -75,7 +75,7 @@ for test_case in "${tests_with_replicated_merge_tree[@]}"; do
 done
 
 # Check for existence of __init__.py files
-for i in "${ROOT_PATH}"/tests/integration/test_*; do FILE="${i}/__init__.py"; [ ! -f "${FILE}" ] && echo "${FILE} should exist for every integration test"; done
+# for i in "${ROOT_PATH}"/tests/integration/test_*; do FILE="${i}/__init__.py"; [ ! -f "${FILE}" ] && echo "${FILE} should exist for every integration test"; done
 
 # Check for executable bit on non-executable files
 find $ROOT_PATH/{src,base,programs,utils,tests,docs,cmake} '(' -name '*.cpp' -or -name '*.h' -or -name '*.sql' -or -name '*.j2' -or -name '*.xml' -or -name '*.reference' -or -name '*.txt' -or -name '*.md' ')' -and -executable | grep -P '.' && echo "These files should not be executable."

docs/en/sql-reference/statements/execute_as.md

@@ -0,0 +1,42 @@
+---
+description: 'Documentation for EXECUTE AS Statement'
+sidebar_label: 'EXECUTE AS'
+sidebar_position: 53
+slug: /sql-reference/statements/execute_as
+title: 'EXECUTE AS Statement'
+doc_type: 'reference'
+---
+
+# EXECUTE AS Statement
+
+Allows to execute queries on behalf of a different user.
+
+## Syntax {#syntax}
+
+```sql
+EXECUTE AS target_user;
+EXECUTE AS target_user subquery;
+```
+
+The first form (without `subquery`) sets that all the following queries in the current session will be executed on behalf of the specified `target_user`.
+
+The second form (with `subquery`) executes only the specified `subquery` on behalf of the specified `target_user`.
+
+In order to work both forms require server setting [allow_impersonate_user](/operations/server-configuration-parameters/settings#allow_impersonate_user)
+to be set to `1` and the `IMPERSONATE` privilege to be granted. For example, the following commands
+```sql
+GRANT IMPERSONATE ON user1 TO user2;
+GRANT IMPERSONATE ON * TO user3;
+```
+allow user `user2` to execute commands `EXECUTE AS user1 ...` and also allow user `user3` to execute commands as any user.
+
+While impersonating another user function [currentUser()](/sql-reference/functions/other-functions#currentUser) returns the name of that other user,
+and function [authenticatedUser()](/sql-reference/functions/other-functions#authenticatedUser) returns the name of the user who has been actually authenticated.
+
+## Examples {#examples}
+
+```sql
+SELECT currentUser(), authenticatedUser(); -- outputs "default    default"
+CREATE USER james;
+EXECUTE AS james SELECT currentUser(), authenticatedUser(); -- outputs "james    default"
+```

src/Access/Common/AccessType.h

@@ -290,6 +290,7 @@ enum class AccessType : uint8_t
     M(SHOW_QUOTAS, "SHOW CREATE QUOTA", GLOBAL, SHOW_ACCESS) \
     M(SHOW_SETTINGS_PROFILES, "SHOW PROFILES, SHOW CREATE SETTINGS PROFILE, SHOW CREATE PROFILE", GLOBAL, SHOW_ACCESS) \
     M(SHOW_ACCESS, "", GROUP, ACCESS_MANAGEMENT) \
+    M(IMPERSONATE, "EXECUTE AS", USER_NAME, ACCESS_MANAGEMENT) \
     M(ACCESS_MANAGEMENT, "", GROUP, ALL) \
     M(SHOW_NAMED_COLLECTIONS, "SHOW NAMED COLLECTIONS", NAMED_COLLECTION, NAMED_COLLECTION_ADMIN) \
     M(SHOW_NAMED_COLLECTIONS_SECRETS, "SHOW NAMED COLLECTIONS SECRETS", NAMED_COLLECTION, NAMED_COLLECTION_ADMIN) \

src/Core/ServerSettings.cpp

@@ -1139,7 +1139,7 @@ The policy on how to perform a scheduling of CPU slots specified by `concurrent_
     DECLARE(UInt64, threadpool_local_fs_reader_queue_size, 1000000, R"(The maximum number of jobs that can be scheduled on the thread pool for reading from local filesystem.)", 0) \
     DECLARE(NonZeroUInt64, threadpool_remote_fs_reader_pool_size, 250, R"(Number of threads in the Thread pool used for reading from remote filesystem when `remote_filesystem_read_method = 'threadpool'`.)", 0) \
     DECLARE(UInt64, threadpool_remote_fs_reader_queue_size, 1000000, R"(The maximum number of jobs that can be scheduled on the thread pool for reading from remote filesystem.)", 0) \
-
+    DECLARE(Bool, allow_impersonate_user, false, R"(Enable/disable the IMPERSONATE feature (EXECUTE AS target_user).)", 0) \
 
 // clang-format on
 

src/Functions/authenticatedUser.cpp

@@ -0,0 +1,83 @@
+#include <Functions/IFunction.h>
+#include <Functions/FunctionFactory.h>
+#include <Interpreters/Context.h>
+#include <DataTypes/DataTypeString.h>
+#include <Core/Field.h>
+
+
+namespace DB
+{
+namespace
+{
+
+class FunctionAuthenticatedUser : public IFunction
+{
+    const String user_name;
+
+public:
+    static constexpr auto name = "authenticatedUser";
+    static FunctionPtr create(ContextPtr context)
+    {
+        return std::make_shared<FunctionAuthenticatedUser>(context->getClientInfo().authenticated_user);
+    }
+
+    explicit FunctionAuthenticatedUser(const String & user_name_) : user_name{user_name_}
+    {
+    }
+
+    String getName() const override
+    {
+        return name;
+    }
+    size_t getNumberOfArguments() const override
+    {
+        return 0;
+    }
+
+    DataTypePtr getReturnTypeImpl(const DataTypes & /*arguments*/) const override
+    {
+        return std::make_shared<DataTypeString>();
+    }
+
+    bool isDeterministic() const override { return false; }
+
+    bool isSuitableForShortCircuitArgumentsExecution(const DataTypesWithConstInfo & /*arguments*/) const override { return false; }
+
+    ColumnPtr executeImpl(const ColumnsWithTypeAndName &, const DataTypePtr &, size_t input_rows_count) const override
+    {
+        return DataTypeString().createColumnConst(input_rows_count, user_name);
+    }
+};
+
+}
+
+REGISTER_FUNCTION(AuthenticatedUser)
+{
+    factory.registerFunction<FunctionAuthenticatedUser>(FunctionDocumentation{
+        .description=R"(
+If the session user has been switched using the EXECUTE AS command, this function returns the name of the original user that was used for authentication and creating the session.
+Alias: authUser()
+        )",
+        .syntax=R"(authenticatedUser())",
+        .arguments={},
+        .returned_value={R"(The name of the authenticated user.)", {"String"}},
+        .examples{
+            {"Usage example",
+            R"(
+            EXECUTE as u1;
+            SELECT currentUser(), authenticatedUser();
+            )",
+            R"(
+┌─currentUser()─┬─authenticatedUser()─┐
+│ u1            │ default             │
+└───────────────┴─────────────────────┘
+        )"
+        }},
+        .introduced_in = {25, 11},
+        .category = FunctionDocumentation::Category::Other
+    });
+
+    factory.registerAlias("authUser", "authenticatedUser");
+}
+
+}

src/Interpreters/Access/InterpreterExecuteAsQuery.cpp

@@ -0,0 +1,118 @@
+#include <Interpreters/Access/InterpreterExecuteAsQuery.h>
+
+#include <Access/AccessControl.h>
+#include <Access/User.h>
+#include <Core/Settings.h>
+#include <Core/ServerSettings.h>
+#include <Parsers/Access/ASTExecuteAsQuery.h>
+#include <Parsers/Access/ASTUserNameWithHost.h>
+#include <Interpreters/Context.h>
+#include <Interpreters/InterpreterFactory.h>
+#include <Interpreters/QueryFlags.h>
+#include <Interpreters/executeQuery.h>
+
+
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int SUPPORT_IS_DISABLED;
+}
+
+namespace ServerSetting
+{
+    extern const ServerSettingsBool allow_impersonate_user;
+}
+
+namespace
+{
+    /// Creates another query context to execute a query as another user.
+    ContextMutablePtr impersonateQueryContext(ContextPtr context, const String & target_user_name)
+    {
+        auto new_context = Context::createCopy(context->getGlobalContext());
+        new_context->setClientInfo(context->getClientInfo());
+        new_context->makeQueryContext();
+
+        const auto & database = context->getCurrentDatabase();
+        if (!database.empty() && database != new_context->getCurrentDatabase())
+            new_context->setCurrentDatabase(database);
+
+        new_context->setInsertionTable(context->getInsertionTable(), context->getInsertionTableColumnNames());
+        new_context->setProgressCallback(context->getProgressCallback());
+        new_context->setProcessListElement(context->getProcessListElement());
+
+        if (context->getCurrentTransaction())
+            new_context->setCurrentTransaction(context->getCurrentTransaction());
+
+        if (context->getZooKeeperMetadataTransaction())
+            new_context->initZooKeeperMetadataTransaction(context->getZooKeeperMetadataTransaction());
+
+        new_context->setUser(context->getAccessControl().getID<User>(target_user_name));
+
+        /// We need to update the client info to make currentUser() return `target_user_name`.
+        new_context->setCurrentUserName(target_user_name);
+        new_context->setInitialUserName(target_user_name);
+
+        auto changed_settings = context->getSettingsRef().changes();
+        new_context->clampToSettingsConstraints(changed_settings, SettingSource::QUERY);
+        new_context->applySettingsChanges(changed_settings);
+
+        return new_context;
+    }
+
+    /// Changes the session context to execute all following queries in this session as another user.
+    void impersonateSessionContext(ContextMutablePtr context, const String & target_user_name)
+    {
+        auto database = context->getCurrentDatabase();
+        auto changed_settings = context->getSettingsRef().changes();
+
+        context->setUser(context->getAccessControl().getID<User>(target_user_name));
+
+        /// We need to update the client info to make currentUser() return `target_user_name`.
+        context->setCurrentUserName(target_user_name);
+        context->setInitialUserName(target_user_name);
+
+        context->clampToSettingsConstraints(changed_settings, SettingSource::QUERY);
+        context->applySettingsChanges(changed_settings);
+
+        if (!database.empty() && database != context->getCurrentDatabase())
+            context->setCurrentDatabase(database);
+    }
+}
+
+
+BlockIO InterpreterExecuteAsQuery::execute()
+{
+    if (!getContext()->getGlobalContext()->getServerSettings()[ServerSetting::allow_impersonate_user])
+        throw Exception(ErrorCodes::SUPPORT_IS_DISABLED, "IMPERSONATE feature is disabled, set allow_impersonate_user to 1 to enable");
+
+    const auto & query = query_ptr->as<const ASTExecuteAsQuery &>();
+    String target_user_name = query.target_user->toString();
+    getContext()->checkAccess(AccessType::IMPERSONATE, target_user_name);
+
+    if (query.subquery)
+    {
+        /// EXECUTE AS <user> <subquery>
+        auto subquery_context = impersonateQueryContext(getContext(), target_user_name);
+        return executeQuery(query.subquery->formatWithSecretsOneLine(), subquery_context, QueryFlags{ .internal = true }).second;
+    }
+    else
+    {
+        /// EXECUTE AS <user>
+        impersonateSessionContext(getContext()->getSessionContext(), target_user_name);
+        return {};
+    }
+}
+
+
+void registerInterpreterExecuteAsQuery(InterpreterFactory & factory)
+{
+    auto create_fn = [] (const InterpreterFactory::Arguments & args)
+    {
+        return std::make_unique<InterpreterExecuteAsQuery>(args.query, args.context);
+    };
+    factory.registerInterpreter("InterpreterExecuteAsQuery", create_fn);
+}
+
+}

src/Interpreters/Access/InterpreterExecuteAsQuery.h

@@ -0,0 +1,20 @@
+#pragma once
+
+#include <Interpreters/IInterpreter.h>
+#include <Parsers/IAST_fwd.h>
+
+
+namespace DB
+{
+
+class InterpreterExecuteAsQuery : public IInterpreter, WithMutableContext
+{
+public:
+    InterpreterExecuteAsQuery(const ASTPtr & query_ptr_, ContextMutablePtr context_) : WithMutableContext(context_), query_ptr(query_ptr_) {}
+    BlockIO execute() override;
+
+private:
+    ASTPtr query_ptr;
+};
+
+}

src/Interpreters/ClientInfo.h

@@ -65,6 +65,9 @@ class ClientInfo
     String current_query_id;
     std::shared_ptr<Poco::Net::SocketAddress> current_address;
 
+    /// For IMPERSONATEd session, stores the original authenticated user
+    String authenticated_user;
+
     /// When query_kind == INITIAL_QUERY, these values are equal to current.
     String initial_user;
     String initial_query_id;

src/Interpreters/DDLWorker.cpp

@@ -1139,7 +1139,6 @@ bool DDLWorker::initializeMainThread()
             auto zookeeper = getAndSetZooKeeper();
             zookeeper->createAncestors(fs::path(queue_dir) / "");
             initializeReplication();
-            markReplicasActive(true);
             initialized = true;
             return true;
         }
@@ -1212,6 +1211,14 @@ void DDLWorker::runMainThread()
             }
 
             cleanup_event->set();
+            try
+            {
+                markReplicasActive(reinitialized);
+            }
+            catch (...)
+            {
+                tryLogCurrentException(log, "An error occurred when markReplicasActive: ");
+            }
             scheduleTasks(reinitialized);
             subsequent_errors_count = 0;
 
@@ -1290,20 +1297,23 @@ void DDLWorker::createReplicaDirs(const ZooKeeperPtr & zookeeper, const NameSet
         zookeeper->createAncestors(fs::path(replicas_dir) / host_id / "");
 }
 
-void DDLWorker::markReplicasActive(bool /*reinitialized*/)
+void DDLWorker::markReplicasActive(bool reinitialized)
 {
     auto zookeeper = getZooKeeper();
 
-    // Reset all active_node_holders
-    for (auto & it : active_node_holders)
+    if (reinitialized)
     {
-        auto & active_node_holder = it.second.second;
-        if (active_node_holder)
-            active_node_holder->setAlreadyRemoved();
-        active_node_holder.reset();
-    }
+        // Reset all active_node_holders
+        for (auto & it : active_node_holders)
+        {
+            auto & active_node_holder = it.second.second;
+            if (active_node_holder)
+                active_node_holder->setAlreadyRemoved();
+            active_node_holder.reset();
+        }
 
-    active_node_holders.clear();
+        active_node_holders.clear();
+    }
 
     for (auto it = active_node_holders.begin(); it != active_node_holders.end();)
     {
@@ -1384,12 +1394,7 @@ void DDLWorker::markReplicasActive(bool /*reinitialized*/)
         {
             zookeeper->deleteEphemeralNodeIfContentMatches(active_path, active_id);
         }
-        Coordination::Requests ops;
-        ops.emplace_back(zkutil::makeCreateRequest(active_path, active_id, zkutil::CreateMode::Ephemeral));
-        /// To bump node mtime
-        ops.emplace_back(zkutil::makeSetRequest(fs::path(replicas_dir) / host_id, "", -1));
-        zookeeper->multi(ops);
-
+        zookeeper->create(active_path, active_id, zkutil::CreateMode::Ephemeral);
         auto active_node_holder_zookeeper = zookeeper;
         auto active_node_holder = zkutil::EphemeralNodeHolder::existing(active_path, *active_node_holder_zookeeper);
         active_node_holders[host_id] = {active_node_holder_zookeeper, active_node_holder};