[#5966] improvment(authorization): Add path based securable object and user group mapping interface

authorizations/authorization-common/src/main/java/org/apache/gravitino/authorization/common/AuthorizationUserGroupMappingProvider.java

@@ -0,0 +1,64 @@
+/*
+ *  Licensed to the Apache Software Foundation (ASF) under one
+ *  or more contributor license agreements.  See the NOTICE file
+ *  distributed with this work for additional information
+ *  regarding copyright ownership.  The ASF licenses this file
+ *  to you under the Apache License, Version 2.0 (the
+ *  "License"); you may not use this file except in compliance
+ *  with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing,
+ *  software distributed under the License is distributed on an
+ *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ *  KIND, either express or implied.  See the License for the
+ *  specific language governing permissions and limitations
+ *  under the License.
+ *
+ */
+
+package org.apache.gravitino.authorization.common;
+
+import java.util.Map;
+
+/**
+ * The AuthorizationUserGroupMappingProvider interface defines the public API for mapping Gravitino
+ * users and groups to the that in underlying data source system.
+ *
+ * <p>Typically, the users and group names in Gravitino are the same as the underlying data source.
+ * However, in some cases, the user and group names in Gravitino may be different from the
+ * underlying data source. For instance, in GCP IAM, the username is the email address or the
+ * service account. So the user group mapping provider can be used to map the Gravitino username to
+ * the email address or service account.
+ */
+public interface AuthorizationUserGroupMappingProvider {
+
+  /**
+   * Initialize the user group mapping provider with the configuration.
+   *
+   * @param config The configuration map for the user group mapping provider.
+   */
+  default void initialize(Map<String, String> config) {}
+
+  /**
+   * Get the username from the underlying data source based on the Gravitino username For instance,
+   * in GCP IAM, the username is the email address or the service account.
+   *
+   * @param gravitinoUserName The Gravitino username.
+   * @return The username from the underlying data source.
+   */
+  default String getUserName(String gravitinoUserName) {
+    return gravitinoUserName;
+  }
+
+  /**
+   * Get the group name from the underlying data source based on the Gravitino group name.
+   *
+   * @param gravitinoGroupName The Gravitino group name.
+   * @return The group name from the underlying data source.
+   */
+  default String getGroupName(String gravitinoGroupName) {
+    return gravitinoGroupName;
+  }
+}

authorizations/authorization-common/src/main/java/org/apache/gravitino/authorization/common/PathBasedMetadataObject.java

@@ -32,7 +32,11 @@ public class PathBasedMetadataObject implements AuthorizationMetadataObject {
    */
   public enum Type implements AuthorizationMetadataObject.Type {
     /** A path is mapped the path of storages like HDFS, S3 etc. */
-    PATH(MetadataObject.Type.FILESET);
+    FILESET_PATH(MetadataObject.Type.FILESET),
+    TABLE_PATH(MetadataObject.Type.TABLE),
+    SCHEMA_PATH(MetadataObject.Type.SCHEMA),
+    CATALOG_PATH(MetadataObject.Type.CATALOG);
+
     private final MetadataObject.Type metadataType;
 
     Type(MetadataObject.Type type) {
@@ -87,7 +91,7 @@ public void validateAuthorizationMetadataObject() throws IllegalArgumentExceptio
         type != null, "Cannot create a path based metadata object with no type");
 
     Preconditions.checkArgument(
-        type == PathBasedMetadataObject.Type.PATH, "it must be the PATH type");
+        type == PathBasedMetadataObject.Type.FILESET_PATH, "it must be the PATH type");
 
     for (String name : names) {
       Preconditions.checkArgument(

authorizations/authorization-ranger/src/main/java/org/apache/gravitino/authorization/ranger/RangerAuthorizationHDFSPlugin.java

@@ -193,11 +193,11 @@ public List<AuthorizationSecurableObject> translatePrivilege(SecurableObject sec
                         if (locationPath != null && !locationPath.isEmpty()) {
                           PathBasedMetadataObject rangerPathBaseMetadataObject =
                               new PathBasedMetadataObject(
-                                  locationPath, PathBasedMetadataObject.Type.PATH);
+                                  locationPath, PathBasedMetadataObject.Type.FILESET_PATH);
                           rangerSecurableObjects.add(
                               generateAuthorizationSecurableObject(
                                   rangerPathBaseMetadataObject.names(),
-                                  PathBasedMetadataObject.Type.PATH,
+                                  PathBasedMetadataObject.Type.FILESET_PATH,
                                   rangerPrivileges));
                         }
                       }
@@ -206,7 +206,7 @@ public List<AuthorizationSecurableObject> translatePrivilege(SecurableObject sec
                       rangerSecurableObjects.add(
                           generateAuthorizationSecurableObject(
                               translateMetadataObject(securableObject).names(),
-                              PathBasedMetadataObject.Type.PATH,
+                              PathBasedMetadataObject.Type.FILESET_PATH,
                               rangerPrivileges));
                       break;
                     default:
@@ -234,7 +234,7 @@ public List<AuthorizationSecurableObject> translatePrivilege(SecurableObject sec
                       rangerSecurableObjects.add(
                           generateAuthorizationSecurableObject(
                               translateMetadataObject(securableObject).names(),
-                              PathBasedMetadataObject.Type.PATH,
+                              PathBasedMetadataObject.Type.FILESET_PATH,
                               rangerPrivileges));
                       break;
                     default:
@@ -265,7 +265,7 @@ public List<AuthorizationSecurableObject> translateOwner(MetadataObject gravitin
         rangerSecurableObjects.add(
             generateAuthorizationSecurableObject(
                 translateMetadataObject(gravitinoMetadataObject).names(),
-                PathBasedMetadataObject.Type.PATH,
+                PathBasedMetadataObject.Type.FILESET_PATH,
                 ownerMappingRule()));
         break;
       default:
@@ -294,17 +294,17 @@ public AuthorizationMetadataObject translateMetadataObject(MetadataObject metada
       case METALAKE:
       case CATALOG:
         rangerPathBaseMetadataObject =
-            new PathBasedMetadataObject("", PathBasedMetadataObject.Type.PATH);
+            new PathBasedMetadataObject("", PathBasedMetadataObject.Type.FILESET_PATH);
         break;
       case SCHEMA:
         rangerPathBaseMetadataObject =
             new PathBasedMetadataObject(
-                metadataObject.fullName(), PathBasedMetadataObject.Type.PATH);
+                metadataObject.fullName(), PathBasedMetadataObject.Type.FILESET_PATH);
         break;
       case FILESET:
         rangerPathBaseMetadataObject =
             new PathBasedMetadataObject(
-                getLocationPath(metadataObject), PathBasedMetadataObject.Type.PATH);
+                getLocationPath(metadataObject), PathBasedMetadataObject.Type.FILESET_PATH);
         break;
       default:
         throw new AuthorizationPluginException(

authorizations/authorization-ranger/src/test/java/org/apache/gravitino/authorization/ranger/integration/test/RangerAuthorizationHDFSPluginIT.java

@@ -56,19 +56,20 @@ public void testTranslateMetadataObject() {
     MetadataObject metalake =
         MetadataObjects.parse(String.format("metalake1"), MetadataObject.Type.METALAKE);
     Assertions.assertEquals(
-        PathBasedMetadataObject.Type.PATH,
+        PathBasedMetadataObject.Type.FILESET_PATH,
         rangerAuthPlugin.translateMetadataObject(metalake).type());
 
     MetadataObject catalog =
         MetadataObjects.parse(String.format("catalog1"), MetadataObject.Type.CATALOG);
     Assertions.assertEquals(
-        PathBasedMetadataObject.Type.PATH,
+        PathBasedMetadataObject.Type.FILESET_PATH,
         rangerAuthPlugin.translateMetadataObject(catalog).type());
 
     MetadataObject schema =
         MetadataObjects.parse(String.format("catalog1.schema1"), MetadataObject.Type.SCHEMA);
     Assertions.assertEquals(
-        PathBasedMetadataObject.Type.PATH, rangerAuthPlugin.translateMetadataObject(schema).type());
+        PathBasedMetadataObject.Type.FILESET_PATH,
+        rangerAuthPlugin.translateMetadataObject(schema).type());
 
     MetadataObject table =
         MetadataObjects.parse(String.format("catalog1.schema1.tab1"), MetadataObject.Type.TABLE);
@@ -81,7 +82,7 @@ public void testTranslateMetadataObject() {
     AuthorizationMetadataObject rangerFileset = rangerAuthPlugin.translateMetadataObject(fileset);
     Assertions.assertEquals(1, rangerFileset.names().size());
     Assertions.assertEquals("/test", rangerFileset.fullName());
-    Assertions.assertEquals(PathBasedMetadataObject.Type.PATH, rangerFileset.type());
+    Assertions.assertEquals(PathBasedMetadataObject.Type.FILESET_PATH, rangerFileset.type());
   }
 
   @Test
@@ -136,7 +137,8 @@ public void testTranslatePrivilege() {
 
     filesetInFileset1.forEach(
         securableObject -> {
-          Assertions.assertEquals(PathBasedMetadataObject.Type.PATH, securableObject.type());
+          Assertions.assertEquals(
+              PathBasedMetadataObject.Type.FILESET_PATH, securableObject.type());
           Assertions.assertEquals("/test", securableObject.fullName());
           Assertions.assertEquals(2, securableObject.privileges().size());
         });
@@ -165,7 +167,7 @@ public void testTranslateOwner() {
     List<AuthorizationSecurableObject> filesetOwner = rangerAuthPlugin.translateOwner(fileset);
     Assertions.assertEquals(1, filesetOwner.size());
     Assertions.assertEquals("/test", filesetOwner.get(0).fullName());
-    Assertions.assertEquals(PathBasedMetadataObject.Type.PATH, filesetOwner.get(0).type());
+    Assertions.assertEquals(PathBasedMetadataObject.Type.FILESET_PATH, filesetOwner.get(0).type());
     Assertions.assertEquals(3, filesetOwner.get(0).privileges().size());
   }
 }