Add enable_user_name_access_type setting for compatibility.

Author: pufit

src/Access/AccessControl.cpp

@@ -306,6 +306,9 @@ void AccessControl::setupFromMainConfig(const Poco::Util::AbstractConfiguration
     setSettingsConstraintsReplacePrevious(config_.getBool("access_control_improvements.settings_constraints_replace_previous", true));
     setTableEnginesRequireGrant(config_.getBool("access_control_improvements.table_engines_require_grant", false));
 
+    /// Set `true` by default because the feature is backward incompatible only when older version replicas are in the same cluster.
+    setEnableUserNameAccessType(config_.getBool("access_control_improvements.enable_user_name_access_type", true));
+
     addStoragesFromMainConfig(config_, config_path_, get_zookeeper_function_);
 
     role_cache = std::make_unique<RoleCache>(*this, config_.getInt("access_control_improvements.role_cache_expiration_time_seconds", 600));
@@ -771,6 +774,15 @@ int AccessControl::getBcryptWorkfactor() const
     return bcrypt_workfactor;
 }
 
+void AccessControl::setEnableUserNameAccessType(bool enable_user_name_access_type_)
+{
+    enable_user_name_access_type = enable_user_name_access_type_;
+}
+
+bool AccessControl::isEnabledUserNameAccessType() const
+{
+    return enable_user_name_access_type;
+}
 
 std::shared_ptr<const ContextAccess> AccessControl::getContextAccess(const ContextAccessParams & params) const
 {

src/Access/AccessControl.h

@@ -173,6 +173,10 @@ class AccessControl : public MultipleAccessStorage
     void setBcryptWorkfactor(int workfactor_);
     int getBcryptWorkfactor() const;
 
+    /// Compatability setting
+    void setEnableUserNameAccessType(bool enable_user_name_access_type_);
+    bool isEnabledUserNameAccessType() const;
+
     /// Enables logic that users without permissive row policies can still read rows using a SELECT query.
     /// For example, if there are two users A, B and a row policy is defined only for A, then
     /// if this setting is true the user B will see all rows, and if this setting is false the user B will see no rows.
@@ -284,6 +288,7 @@ class AccessControl : public MultipleAccessStorage
     std::atomic<AuthenticationType> default_password_type = AuthenticationType::SHA256_PASSWORD;
     std::atomic_bool allow_experimental_tier_settings = true;
     std::atomic_bool allow_beta_tier_settings = true;
+    std::atomic_bool enable_user_name_access_type = true;
 };
 
 }

src/Access/Common/AccessRightsElement.cpp

@@ -1,7 +1,9 @@
+#include <Access/AccessControl.h>
 #include <Access/Common/AccessRightsElement.h>
 #include <Common/quoteString.h>
 #include <IO/Operators.h>
 #include <IO/WriteBufferFromString.h>
+#include <Interpreters/Context.h>
 #include <Parsers/IAST.h>
 
 
@@ -139,16 +141,36 @@ void AccessRightsElement::formatColumnNames(WriteBuffer & buffer) const
 
 void AccessRightsElement::formatONClause(WriteBuffer & buffer, bool hilite) const
 {
+    const auto context = Context::getGlobalContextInstance();
+    const auto & access_control = context->getAccessControl();
+
     buffer << (hilite ? IAST::hilite_keyword : "") << "ON " << (hilite ? IAST::hilite_none : "");
     if (isGlobalWithParameter())
     {
-        if (anyParameter())
-            buffer << "*";
+        /// Special check for backward compatibility.
+        /// If `enable_user_name_access_type` is set to false, we will dump `GRANT CREATE USER ON *` as `GRANT CREATE USER ON *.*`.
+        /// This will allow us to run old replicas in the same cluster.
+        if (access_flags.getParameterType() == AccessFlags::USER_NAME
+            && !access_control.isEnabledUserNameAccessType())
+        {
+            if (!anyParameter())
+                getLogger("AccessRightsElement")->warning(
+                    "Converting {} to *.* because the setting `enable_user_name_access_type` is `false`. "
+                    "Consider turning this setting on, if your cluster contains no replicas older than 25.1",
+                    parameter);
+
+            buffer << "*.*";
+        }
         else
         {
-            buffer << backQuoteIfNeed(parameter);
-            if (wildcard)
+            if (anyParameter())
                 buffer << "*";
+            else
+            {
+                buffer << backQuoteIfNeed(parameter);
+                if (wildcard)
+                    buffer << "*";
+            }
         }
     }
     else if (anyDatabase())